Bases: TelemetrySettingsProvider[MsgType, KafkaPublishCommand]
Source code in faststream/confluent/opentelemetry/provider.py
| def __init__(self) -> None:
self.messaging_system = "kafka"
|
messaging_system instance-attribute
messaging_system = 'kafka'
get_publish_attrs_from_cmd
Source code in faststream/confluent/opentelemetry/provider.py
27
28
29
30
31
32
33
34
35
36
37
38
39
40 | def get_publish_attrs_from_cmd(self, cmd: "KafkaPublishCommand") -> dict[str, Any]:
attrs: dict[str, Any] = {
SpanAttributes.MESSAGING_SYSTEM: self.messaging_system,
SpanAttributes.MESSAGING_DESTINATION_NAME: cmd.destination,
SpanAttributes.MESSAGING_MESSAGE_CONVERSATION_ID: cmd.correlation_id,
}
if cmd.partition is not None:
attrs[SpanAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION] = cmd.partition
if cmd.key is not None:
attrs[SpanAttributes.MESSAGING_KAFKA_MESSAGE_KEY] = cmd.key
return attrs
|
get_publish_destination_name
Source code in faststream/confluent/opentelemetry/provider.py
| @override
def get_publish_destination_name(self, cmd: "PublishCommand") -> str:
return cmd.destination
|
get_consume_attrs_from_message
get_consume_attrs_from_message(
msg: StreamMessage[MsgType],
) -> dict[str, Any]
Source code in faststream/opentelemetry/provider.py
| def get_consume_attrs_from_message(
self,
msg: "StreamMessage[MsgType]",
) -> dict[str, Any]: ...
|
get_consume_destination_name
get_consume_destination_name(
msg: StreamMessage[MsgType],
) -> str
Source code in faststream/opentelemetry/provider.py
| def get_consume_destination_name(
self,
msg: "StreamMessage[MsgType]",
) -> str: ...
|