Add message queue time metric (#37)

metaplex-foundation · Feb 2, 2023 · 5e32f46 · 5e32f46
1 parent 7fa24a8
commit 5e32f46
Showing 5 changed files with 18 additions and 8 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/plerkle/Cargo.toml b/plerkle/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "plerkle"
 description = "Geyser plugin with dynamic config reloading, message bus agnostic abstractions and a whole lot of fun."
-version = "1.3.1"
+version = "1.3.2"
 authors = ["Metaplex Developers <dev@metaplex.com>"]
 repository = "https://github.com/metaplex-foundation/digital-asset-validator-plugin"
 license = "AGPL-3.0"
@@ -31,9 +31,9 @@ cadence-macros = "0.29.0"
 chrono = "0.4.19"
 tracing = "0.1.35"
 hex = "0.4.3"
-plerkle_messenger = { path = "../plerkle_messenger", version = "1.3.1", features = ["redis"] }
+plerkle_messenger = { path = "../plerkle_messenger", version = "1.3.2", features = ["redis"] }
 flatbuffers = "22.10.26"
-plerkle_serialization = { path = "../plerkle_serialization", version = "1.3.1" }
+plerkle_serialization = { path = "../plerkle_serialization", version = "1.3.2" }
 tokio = { version = "1.23.0", features = ["full"] }
 figment = { version = "0.10.6", features = ["env", "test"] }
 

diff --git a/plerkle/src/geyser_plugin_nft.rs b/plerkle/src/geyser_plugin_nft.rs
@@ -38,6 +38,7 @@ use tokio::{
 struct SerializedData<'a> {
     stream: &'static str,
     builder: FlatBufferBuilder<'a>,
+    seen_at: Instant
 }
 
 #[derive(Default)]
@@ -254,6 +255,7 @@ impl GeyserPlugin for Plerkle<'static> {
                     let bytes = data.builder.finished_data();
                     let _ = messenger.send(data.stream, bytes).await;
                     safe_metric(|| {
+                        statsd_time!("message_send_queue_time", data.seen_at.elapsed());
                         statsd_time!("message_send_latency", start.elapsed());
                     })
                 }
@@ -273,6 +275,7 @@ impl GeyserPlugin for Plerkle<'static> {
         slot: u64,
         is_startup: bool,
     ) -> solana_geyser_plugin_interface::geyser_plugin_interface::Result<()> {
+        let seen = Instant::now();
         if !self.handle_startup && is_startup {
             return Ok(());
         }
@@ -315,6 +318,7 @@ impl GeyserPlugin for Plerkle<'static> {
             let data = SerializedData {
                 stream: ACCOUNT_STREAM,
                 builder,
+                seen_at: seen,
             };
             let _ = sender.send(data).await;
             safe_metric(|| {
@@ -341,6 +345,7 @@ impl GeyserPlugin for Plerkle<'static> {
         parent: Option<u64>,
         status: SlotStatus,
     ) -> solana_geyser_plugin_interface::geyser_plugin_interface::Result<()> {
+        let seen = Instant::now();
         // Get runtime and sender channel.
         let runtime = self.get_runtime()?;
         let sender = self.get_sender_clone()?;
@@ -354,6 +359,7 @@ impl GeyserPlugin for Plerkle<'static> {
             let data = SerializedData {
                 stream: SLOT_STREAM,
                 builder,
+                seen_at: seen,
             };
             let _ = sender.send(data).await;
         });
@@ -366,6 +372,7 @@ impl GeyserPlugin for Plerkle<'static> {
         transaction_info: ReplicaTransactionInfoVersions,
         slot: u64,
     ) -> solana_geyser_plugin_interface::geyser_plugin_interface::Result<()> {
+        let seen = Instant::now();
         let rep: ReplicaTransactionInfoV2;
         let transaction_info = match transaction_info {
             ReplicaTransactionInfoVersions::V0_0_2(ti) => ti,
@@ -409,6 +416,7 @@ impl GeyserPlugin for Plerkle<'static> {
             let data = SerializedData {
                 stream: TRANSACTION_STREAM,
                 builder,
+                seen_at: seen,
             };
             let _ = sender.send(data).await;
         });
@@ -422,6 +430,7 @@ impl GeyserPlugin for Plerkle<'static> {
         &mut self,
         blockinfo: ReplicaBlockInfoVersions,
     ) -> solana_geyser_plugin_interface::geyser_plugin_interface::Result<()> {
+        let seen = Instant::now();
         match blockinfo {
             ReplicaBlockInfoVersions::V0_0_1(block_info) => {
                 // Get runtime and sender channel.
@@ -437,6 +446,7 @@ impl GeyserPlugin for Plerkle<'static> {
                     let data = SerializedData {
                         stream: BLOCK_STREAM,
                         builder,
+                        seen_at: seen
                     };
                     let _ = sender.send(data).await;
                 });

diff --git a/plerkle_messenger/Cargo.toml b/plerkle_messenger/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "plerkle_messenger"
 description = "Metaplex Messenger trait for Geyser plugin producer/consumer patterns."
-version = "1.3.1"
+version = "1.3.2"
 authors = ["Metaplex Developers <dev@metaplex.com>"]
 repository = "https://github.com/metaplex-foundation/digital-asset-validator-plugin"
 license = "AGPL-3.0"

diff --git a/plerkle_serialization/Cargo.toml b/plerkle_serialization/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "plerkle_serialization"
 description = "Metaplex Flatbuffers Plerkle Serialization for Geyser plugin producer/consumer patterns."
-version = "1.3.1"
+version = "1.3.2"
 authors = ["Metaplex Developers <dev@metaplex.com>"]
 repository = "https://github.com/metaplex-foundation/digital-asset-validator-plugin"
 license = "AGPL-3.0"