STR-386: restore missing metric functions dropped during OnceLock conflict resolution

Signed-off-by: GitHub <noreply@github.com>

Author: Ozodimgba

yellowstone-grpc-geyser/src/metrics.rs

@@ -17,7 +17,8 @@ use {
     },
     log::{debug, error, info},
     prometheus::{
-        Histogram, HistogramOpts, IntCounterVec, IntGauge, IntGaugeVec, Opts, Registry, TextEncoder,
+        Histogram, HistogramOpts, IntCounter, IntCounterVec, IntGauge, IntGaugeVec, Opts, Registry,
+        TextEncoder,
     },
     solana_clock::Slot,
     std::{
@@ -188,6 +189,34 @@ static ref FALLBACK_ENCODE_TIME_US: Histogram = Histogram::with_opts(
     )
     .buckets(vec![1.0, 5.0, 10.0, 25.0, 50.0, 100.0, 250.0, 500.0])
 ).unwrap();
+
+static ref GRPC_CONCURRENT_SUBSCRIBE_PER_TCP_CONNECTION: IntGaugeVec = IntGaugeVec::new(
+        Opts::new(
+            "grpc_concurrent_subscribe_per_tcp_connection",
+            "Current concurrent subscriptions per remote TCP peer socket address"
+        ),
+        &["remote_peer_sk_addr"]
+    ).unwrap();
+
+    static ref TOTAL_TRAFFIC_SENT: IntCounter = IntCounter::new(
+        "total_traffic_sent_bytes",
+        "Total traffic sent to subscriber by type (account_update, block_meta, etc)"
+    ).unwrap();
+
+    static ref TRAFFIC_SENT_PER_REMOTE_IP: IntCounterVec = IntCounterVec::new(
+        Opts::new("traffic_sent_per_remote_ip_bytes", "Total traffic sent to subscriber by remote IP"),
+        &["remote_ip"]
+    ).unwrap();
+
+    static ref GRPC_METHOD_CALL_COUNT: IntCounterVec = IntCounterVec::new(
+        Opts::new("yellowstone_grpc_method_call_count", "Total number of calls to GetVersion gRPC method"),
+        &["method"]
+    ).unwrap();
+
+    static ref GRPC_SERVICE_OUTBOUND_BYTES: IntGaugeVec = IntGaugeVec::new(
+        Opts::new("yellowstone_grpc_service_outbound_bytes", "Current emitted bytes by tonic service response bodies per active subscriber stream"),
+        &["subscriber_id"]
+    ).unwrap();
 }
 
 #[derive(Debug)]
@@ -350,6 +379,11 @@ impl PrometheusService {
             register!(PARALLEL_ENCODER_QUEUE_DEPTH);
             register!(FAST_PATH_TIME_US);
             register!(FALLBACK_ENCODE_TIME_US);
+            register!(GRPC_CONCURRENT_SUBSCRIBE_PER_TCP_CONNECTION);
+            register!(TOTAL_TRAFFIC_SENT);
+            register!(TRAFFIC_SENT_PER_REMOTE_IP);
+            register!(GRPC_METHOD_CALL_COUNT);
+            register!(GRPC_SERVICE_OUTBOUND_BYTES);
 
             VERSION
                 .with_label_values(&[
@@ -612,6 +646,58 @@ pub fn observe_fallback_encode_time_us(latency: f64) {
     FALLBACK_ENCODE_TIME_US.observe(latency);
 }
 
+pub fn incr_grpc_method_call_count<S: AsRef<str>>(method: S) {
+    GRPC_METHOD_CALL_COUNT
+        .with_label_values(&[method.as_ref()])
+        .inc();
+}
+
+pub fn add_grpc_service_outbound_bytes<S: AsRef<str>>(subscriber_id: S, bytes: u64) {
+    GRPC_SERVICE_OUTBOUND_BYTES
+        .with_label_values(&[subscriber_id.as_ref()])
+        .add(bytes as i64);
+}
+
+pub fn reset_grpc_service_outbound_bytes<S: AsRef<str>>(subscriber_id: S) {
+    GRPC_SERVICE_OUTBOUND_BYTES
+        .with_label_values(&[subscriber_id.as_ref()])
+        .set(0);
+}
+
+pub fn add_total_traffic_sent(bytes: u64) {
+    TOTAL_TRAFFIC_SENT.inc_by(bytes);
+}
+
+pub fn add_traffic_sent_per_remote_ip<S: AsRef<str>>(remote_ip: S, bytes: u64) {
+    TRAFFIC_SENT_PER_REMOTE_IP
+        .with_label_values(&[remote_ip.as_ref()])
+        .inc_by(bytes);
+}
+
+pub fn reset_traffic_sent_per_remote_ip<S: AsRef<str>>(remote_ip: S) {
+    TRAFFIC_SENT_PER_REMOTE_IP
+        .with_label_values(&[remote_ip.as_ref()])
+        .reset();
+    TRAFFIC_SENT_PER_REMOTE_IP
+        .remove_label_values(&[remote_ip.as_ref()])
+        .expect("remove_label_values");
+}
+
+pub fn set_grpc_concurrent_subscribe_per_tcp_connection<S: AsRef<str>>(
+    remote_peer_sk_addr: S,
+    size: u64,
+) {
+    GRPC_CONCURRENT_SUBSCRIBE_PER_TCP_CONNECTION
+        .with_label_values(&[remote_peer_sk_addr.as_ref()])
+        .set(size as i64);
+}
+
+pub fn remove_grpc_concurrent_subscribe_per_tcp_connection<S: AsRef<str>>(remote_peer_sk_addr: S) {
+    GRPC_CONCURRENT_SUBSCRIBE_PER_TCP_CONNECTION
+        .remove_label_values(&[remote_peer_sk_addr.as_ref()])
+        .expect("remove_label_values");
+}
+
 /// Reset all metrics on plugin unload to prevent metric accumulation across plugin lifecycle
 pub fn reset_metrics() {
     // Reset gauge metrics to 0
@@ -632,6 +718,11 @@ pub fn reset_metrics() {
     GRPC_MESSAGE_SENT.reset();
     GRPC_BYTES_SENT.reset();
     GRPC_CLIENT_DISCONNECTS.reset();
+    GRPC_CONCURRENT_SUBSCRIBE_PER_TCP_CONNECTION.reset();
+    TOTAL_TRAFFIC_SENT.reset();
+    TRAFFIC_SENT_PER_REMOTE_IP.reset();
+    GRPC_SERVICE_OUTBOUND_BYTES.reset();
+    GRPC_METHOD_CALL_COUNT.reset();
 
     // Note: VERSION and GEYSER_ACCOUNT_UPDATE_RECEIVED are intentionally not reset
     // - VERSION contains build info set once on startup

yellowstone-grpc-geyser/src/plugin/filter/message.rs

@@ -1,18 +1,24 @@
 use {
-    crate::{metrics, plugin::{
-        filter::{FilterAccountsDataSlice, name::FilterName},
-        message::{
-            MessageAccount, MessageAccountInfo, MessageBlock, MessageBlockMeta, MessageEntry,
-            MessageSlot, MessageTransaction, MessageTransactionInfo,
+    crate::{
+        metrics,
+        plugin::{
+            filter::{name::FilterName, FilterAccountsDataSlice},
+            message::{
+                MessageAccount, MessageAccountInfo, MessageBlock, MessageBlockMeta, MessageEntry,
+                MessageSlot, MessageTransaction, MessageTransactionInfo,
+            },
         },
-    }},
+    },
     bytes::{
-        Bytes, buf::{Buf, BufMut}
+        buf::{Buf, BufMut},
+        Bytes,
     },
     prost::{
-        DecodeError, encoding::{
-            DecodeContext, WireType, encode_key, encode_varint, encoded_len_varint, key_len, message
-        }
+        encoding::{
+            encode_key, encode_varint, encoded_len_varint, key_len, message, DecodeContext,
+            WireType,
+        },
+        DecodeError,
     },
     prost_types::Timestamp,
     smallvec::SmallVec,
@@ -25,7 +31,11 @@ use {
     },
     yellowstone_grpc_proto::{
         geyser::{
-            SlotStatus as SlotStatusProto, SubscribeUpdate, SubscribeUpdateAccount, SubscribeUpdateAccountInfo, SubscribeUpdateBlock, SubscribeUpdateEntry, SubscribeUpdatePing, SubscribeUpdatePong, SubscribeUpdateSlot, SubscribeUpdateTransaction, SubscribeUpdateTransactionInfo, SubscribeUpdateTransactionStatus, subscribe_update::UpdateOneof
+            subscribe_update::UpdateOneof, SlotStatus as SlotStatusProto, SubscribeUpdate,
+            SubscribeUpdateAccount, SubscribeUpdateAccountInfo, SubscribeUpdateBlock,
+            SubscribeUpdateEntry, SubscribeUpdatePing, SubscribeUpdatePong, SubscribeUpdateSlot,
+            SubscribeUpdateTransaction, SubscribeUpdateTransactionInfo,
+            SubscribeUpdateTransactionStatus,
         },
         solana::storage::confirmed_block,
     },