Add tracing spans for sending queue messages (#86)

svix · May 17, 2024 · 62ca8fa · 62ca8fa
1 parent 94bf7bb
commit 62ca8fa
Show file tree

Hide file tree

Showing 5 changed files with 41 additions and 0 deletions.
diff --git a/omniqueue/src/backends/azure_queue_storage.rs b/omniqueue/src/backends/azure_queue_storage.rs
@@ -103,6 +103,14 @@ impl AqsProducer {
         self.send_raw_scheduled(payload, Duration::ZERO).await
     }
 
+    #[tracing::instrument(
+        name = "send",
+        skip_all,
+        fields(
+            payload_size = payload.len(),
+            delay = (delay > Duration::ZERO).then(|| tracing::field::debug(delay))
+        )
+    )]
     pub async fn send_raw_scheduled(&self, payload: &str, delay: Duration) -> Result<()> {
         self.client
             .put_message(payload)

diff --git a/omniqueue/src/backends/gcp_pubsub.rs b/omniqueue/src/backends/gcp_pubsub.rs
@@ -126,6 +126,11 @@ impl GcpPubSubProducer {
         })
     }
 
+    #[tracing::instrument(
+        name = "send",
+        skip_all,
+        fields(payload_size = payload.len())
+    )]
     pub async fn send_raw(&self, payload: &[u8]) -> Result<()> {
         let msg = PubsubMessage {
             data: payload.to_vec(),

diff --git a/omniqueue/src/backends/rabbitmq.rs b/omniqueue/src/backends/rabbitmq.rs
@@ -164,6 +164,11 @@ impl RabbitMqProducer {
         Ok(())
     }
 
+    #[tracing::instrument(
+        name = "send",
+        skip_all,
+        fields(payload_size = payload.len())
+    )]
     pub async fn send_raw(&self, payload: &[u8]) -> Result<()> {
         self.send_raw_with_headers(payload, None).await
     }
@@ -173,6 +178,11 @@ impl RabbitMqProducer {
         self.send_raw(&payload).await
     }
 
+    #[tracing::instrument(
+        name = "send",
+        skip_all,
+        fields(payload_size = payload.len(), delay)
+    )]
     pub async fn send_raw_scheduled(&self, payload: &[u8], delay: Duration) -> Result<()> {
         let mut headers = FieldTable::default();
 

diff --git a/omniqueue/src/backends/redis/mod.rs b/omniqueue/src/backends/redis/mod.rs
@@ -533,6 +533,11 @@ pub struct RedisProducer<M: ManageConnection> {
 }
 
 impl<R: RedisConnection> RedisProducer<R> {
+    #[tracing::instrument(
+        name = "send",
+        skip_all,
+        fields(payload_size = payload.len())
+    )]
     pub async fn send_raw(&self, payload: &[u8]) -> Result<()> {
         if self.use_redis_streams {
             streams::send_raw(self, payload).await
@@ -546,6 +551,11 @@ impl<R: RedisConnection> RedisProducer<R> {
         self.send_raw(&payload).await
     }
 
+    #[tracing::instrument(
+        name = "send",
+        skip_all,
+        fields(payload_size = payload.len(), delay)
+    )]
     pub async fn send_raw_scheduled(&self, payload: &[u8], delay: Duration) -> Result<()> {
         let timestamp = unix_timestamp(SystemTime::now() + delay).map_err(QueueError::generic)?;
 

diff --git a/omniqueue/src/backends/sqs.rs b/omniqueue/src/backends/sqs.rs
@@ -240,6 +240,14 @@ impl SqsProducer {
         self.send_raw(&payload).await
     }
 
+    #[tracing::instrument(
+        name = "send",
+        skip_all,
+        fields(
+            payload_size = payload.len(),
+            delay = (delay > Duration::ZERO).then(|| tracing::field::debug(delay))
+        )
+    )]
     pub async fn send_raw_scheduled(&self, payload: &str, delay: Duration) -> Result<()> {
         if payload.len() > MAX_PAYLOAD_SIZE {
             return Err(QueueError::PayloadTooLarge {