feat: count Redis consuming batch size

airtai · May 11, 2024 · 7aa6d44 · 7aa6d44
1 parent 0eecd2f
commit 7aa6d44
Showing 1 changed file with 6 additions and 1 deletion.
diff --git a/faststream/redis/opentelemetry/provider.py b/faststream/redis/opentelemetry/provider.py
@@ -20,14 +20,19 @@ def get_consume_attrs_from_message(
         self,
         msg: "StreamMessage[AnyDict]",
     ) -> "AnyDict":
-        return {
+        attrs = {
             SpanAttributes.MESSAGING_SYSTEM: self.messaging_system,
             SpanAttributes.MESSAGING_MESSAGE_ID: msg.message_id,
             SpanAttributes.MESSAGING_MESSAGE_CONVERSATION_ID: msg.correlation_id,
             SpanAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES: len(msg.body),
             MESSAGING_DESTINATION_PUBLISH_NAME: msg.raw_message["channel"],
         }
 
+        if msg.raw_message.get("type", "").startswith("b"):
+            attrs[SpanAttributes.MESSAGING_BATCH_MESSAGE_COUNT] = len(msg.decoded_body)
+
+        return attrs
+
     def get_consume_destination_name(
         self,
         msg: "StreamMessage[AnyDict]",