Bases: TelemetrySettingsProvider['AnyDict']
Source code in faststream/redis/opentelemetry/provider.py
| def __init__(self) -> None:
self.messaging_system = "redis"
|
messaging_system instance-attribute
messaging_system = 'redis'
get_consume_attrs_from_message
get_consume_attrs_from_message(msg)
Source code in faststream/redis/opentelemetry/provider.py
| def get_consume_attrs_from_message(
self,
msg: "StreamMessage[AnyDict]",
) -> "AnyDict":
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 cast(str, msg.raw_message.get("type", "")).startswith("b"):
attrs[SpanAttributes.MESSAGING_BATCH_MESSAGE_COUNT] = len(
cast(Sized, msg._decoded_body)
)
return attrs
|
get_consume_destination_name
get_consume_destination_name(msg)
Source code in faststream/redis/opentelemetry/provider.py
| def get_consume_destination_name(
self,
msg: "StreamMessage[AnyDict]",
) -> str:
return self._get_destination(msg.raw_message)
|
get_publish_attrs_from_kwargs
get_publish_attrs_from_kwargs(kwargs)
Source code in faststream/redis/opentelemetry/provider.py
| def get_publish_attrs_from_kwargs(
self,
kwargs: "AnyDict",
) -> "AnyDict":
return {
SpanAttributes.MESSAGING_SYSTEM: self.messaging_system,
SpanAttributes.MESSAGING_DESTINATION_NAME: self._get_destination(kwargs),
SpanAttributes.MESSAGING_MESSAGE_CONVERSATION_ID: kwargs["correlation_id"],
}
|
get_publish_destination_name
get_publish_destination_name(kwargs)
Source code in faststream/redis/opentelemetry/provider.py
| def get_publish_destination_name(
self,
kwargs: "AnyDict",
) -> str:
return self._get_destination(kwargs)
|