|
@@ -531,7 +531,8 @@ start_outgoing_trace(Packet, Attrs, ParentCtx) ->
|
|
|
#{attributes => Attrs}
|
|
#{attributes => Attrs}
|
|
|
),
|
|
),
|
|
|
NCtx = otel_tracer:set_current_span(ParentCtx, SpanCtx),
|
|
NCtx = otel_tracer:set_current_span(ParentCtx, SpanCtx),
|
|
|
- _NPacketWithCtx = put_ctx(NCtx, Packet).
|
|
|
|
|
|
|
+ Packet1WithCtx = put_ctx(NCtx, Packet),
|
|
|
|
|
+ _NPacketWithAttrsg = put_attrs(Attrs, Packet1WithCtx).
|
|
|
|
|
|
|
|
%% Ctx attached in Delivers in `broker_publish/2` and
|
|
%% Ctx attached in Delivers in `broker_publish/2` and
|
|
|
%% transformed to Packet when outgoing
|
|
%% transformed to Packet when outgoing
|
|
@@ -566,9 +567,9 @@ end_span(_) ->
|
|
|
outgoing_maybe_awaiting_next(?PACKET(?PUBACK)) ->
|
|
outgoing_maybe_awaiting_next(?PACKET(?PUBACK)) ->
|
|
|
%% PUBACK (QoS=1), Ignore
|
|
%% PUBACK (QoS=1), Ignore
|
|
|
ok;
|
|
ok;
|
|
|
-outgoing_maybe_awaiting_next(?PUBREC_PACKET(PacketId) = OutgoingPubrecPacket) ->
|
|
|
|
|
|
|
+outgoing_maybe_awaiting_next(?PUBREC_PACKET(PacketId) = Packet) ->
|
|
|
%% PUBREC (QoS=2), Awaiting PUBREL
|
|
%% PUBREC (QoS=2), Awaiting PUBREL
|
|
|
- start_awaiting_trace(?PUBREL, PacketId, get_ctx(OutgoingPubrecPacket));
|
|
|
|
|
|
|
+ start_awaiting_trace(?PUBREL, PacketId, Packet);
|
|
|
outgoing_maybe_awaiting_next(?PACKET(?PUBCOMP)) ->
|
|
outgoing_maybe_awaiting_next(?PACKET(?PUBCOMP)) ->
|
|
|
%% PUBCOMP (QoS=2), Ignore
|
|
%% PUBCOMP (QoS=2), Ignore
|
|
|
ok;
|
|
ok;
|
|
@@ -579,25 +580,27 @@ outgoing_maybe_awaiting_next(?PUBLISH_PACKET(?QOS_0)) ->
|
|
|
ok;
|
|
ok;
|
|
|
outgoing_maybe_awaiting_next(?PUBLISH_PACKET(?QOS_1, PacketId) = Packet) ->
|
|
outgoing_maybe_awaiting_next(?PUBLISH_PACKET(?QOS_1, PacketId) = Packet) ->
|
|
|
%% PUBLISH (QoS=1), Awaiting PUBACK
|
|
%% PUBLISH (QoS=1), Awaiting PUBACK
|
|
|
- start_awaiting_trace(?PUBACK, PacketId, get_ctx(Packet));
|
|
|
|
|
|
|
+ start_awaiting_trace(?PUBACK, PacketId, Packet);
|
|
|
outgoing_maybe_awaiting_next(?PUBLISH_PACKET(?QOS_2, PacketId) = Packet) ->
|
|
outgoing_maybe_awaiting_next(?PUBLISH_PACKET(?QOS_2, PacketId) = Packet) ->
|
|
|
%% PUBLISH (QoS=2), Awaiting PUBREC
|
|
%% PUBLISH (QoS=2), Awaiting PUBREC
|
|
|
- start_awaiting_trace(?PUBREC, PacketId, get_ctx(Packet));
|
|
|
|
|
|
|
+ start_awaiting_trace(?PUBREC, PacketId, Packet);
|
|
|
outgoing_maybe_awaiting_next(?PACKET(?PUBREL, PacketId) = Packet) ->
|
|
outgoing_maybe_awaiting_next(?PACKET(?PUBREL, PacketId) = Packet) ->
|
|
|
%% PUBREL (QoS=2), Awaiting PUBCOMP
|
|
%% PUBREL (QoS=2), Awaiting PUBCOMP
|
|
|
- start_awaiting_trace(?PUBCOMP, PacketId, get_ctx(Packet));
|
|
|
|
|
|
|
+ start_awaiting_trace(?PUBCOMP, PacketId, Packet);
|
|
|
%% ====================
|
|
%% ====================
|
|
|
outgoing_maybe_awaiting_next(_) ->
|
|
outgoing_maybe_awaiting_next(_) ->
|
|
|
%% TODO: Awaiting AUTH
|
|
%% TODO: Awaiting AUTH
|
|
|
ok.
|
|
ok.
|
|
|
|
|
|
|
|
-start_awaiting_trace(AwaitingType, PacketId, ParentCtx) ->
|
|
|
|
|
|
|
+start_awaiting_trace(AwaitingType, PacketId, Packet) ->
|
|
|
AwaitingCtxKey = internal_extra_key(AwaitingType, PacketId),
|
|
AwaitingCtxKey = internal_extra_key(AwaitingType, PacketId),
|
|
|
|
|
+ ParentCtx = get_ctx(Packet),
|
|
|
|
|
+ Attrs = get_attrs(Packet),
|
|
|
AwaitingSpanCtx = otel_tracer:start_span(
|
|
AwaitingSpanCtx = otel_tracer:start_span(
|
|
|
ParentCtx,
|
|
ParentCtx,
|
|
|
?current_tracer,
|
|
?current_tracer,
|
|
|
awaiting_span_name(AwaitingType),
|
|
awaiting_span_name(AwaitingType),
|
|
|
- #{}
|
|
|
|
|
|
|
+ #{attributes => Attrs}
|
|
|
),
|
|
),
|
|
|
NCtx = otel_tracer:set_current_span(ParentCtx, AwaitingSpanCtx),
|
|
NCtx = otel_tracer:set_current_span(ParentCtx, AwaitingSpanCtx),
|
|
|
_ = attach_internal_ctx(AwaitingCtxKey, NCtx),
|
|
_ = attach_internal_ctx(AwaitingCtxKey, NCtx),
|
|
@@ -764,15 +767,19 @@ put_ctx(OtelCtx, #delivery{message = Msg} = Delivery) ->
|
|
|
%% Trace Context in packet
|
|
%% Trace Context in packet
|
|
|
put_ctx(
|
|
put_ctx(
|
|
|
OtelCtx,
|
|
OtelCtx,
|
|
|
- #mqtt_packet{variable = #mqtt_packet_publish{properties = Props} = PubPacket} = Packet
|
|
|
|
|
|
|
+ #mqtt_packet{
|
|
|
|
|
+ variable = #mqtt_packet_publish{properties = Props} = PubPacket
|
|
|
|
|
+ } = Packet
|
|
|
) ->
|
|
) ->
|
|
|
- NProps = to_properties(OtelCtx, Props),
|
|
|
|
|
|
|
+ NProps = to_properties(?EMQX_OTEL_CTX, OtelCtx, Props),
|
|
|
Packet#mqtt_packet{variable = PubPacket#mqtt_packet_publish{properties = NProps}};
|
|
Packet#mqtt_packet{variable = PubPacket#mqtt_packet_publish{properties = NProps}};
|
|
|
put_ctx(
|
|
put_ctx(
|
|
|
OtelCtx,
|
|
OtelCtx,
|
|
|
- #mqtt_packet{variable = #mqtt_packet_puback{properties = Props} = PubAckPacket} = Packet
|
|
|
|
|
|
|
+ #mqtt_packet{
|
|
|
|
|
+ variable = #mqtt_packet_puback{properties = Props} = PubAckPacket
|
|
|
|
|
+ } = Packet
|
|
|
) ->
|
|
) ->
|
|
|
- NProps = to_properties(OtelCtx, Props),
|
|
|
|
|
|
|
+ NProps = to_properties(?EMQX_OTEL_CTX, OtelCtx, Props),
|
|
|
Packet#mqtt_packet{variable = PubAckPacket#mqtt_packet_puback{properties = NProps}};
|
|
Packet#mqtt_packet{variable = PubAckPacket#mqtt_packet_puback{properties = NProps}};
|
|
|
put_ctx(
|
|
put_ctx(
|
|
|
_OtelCtx,
|
|
_OtelCtx,
|
|
@@ -781,24 +788,60 @@ put_ctx(
|
|
|
Any.
|
|
Any.
|
|
|
|
|
|
|
|
get_ctx(#message{extra = Extra}) ->
|
|
get_ctx(#message{extra = Extra}) ->
|
|
|
- from_extra(Extra);
|
|
|
|
|
|
|
+ from_extra(?EMQX_OTEL_CTX, Extra);
|
|
|
get_ctx(#delivery{message = #message{extra = Extra}}) ->
|
|
get_ctx(#delivery{message = #message{extra = Extra}}) ->
|
|
|
- from_extra(Extra);
|
|
|
|
|
|
|
+ from_extra(?EMQX_OTEL_CTX, Extra);
|
|
|
get_ctx(#mqtt_packet{
|
|
get_ctx(#mqtt_packet{
|
|
|
variable = PktVar
|
|
variable = PktVar
|
|
|
}) when is_tuple(PktVar) ->
|
|
}) when is_tuple(PktVar) ->
|
|
|
- from_extra(maps:get(?MQTT_INTERNAL_EXTRA, emqx_packet:info(properties, PktVar), #{}));
|
|
|
|
|
|
|
+ from_extra(
|
|
|
|
|
+ ?EMQX_OTEL_CTX,
|
|
|
|
|
+ maps:get(
|
|
|
|
|
+ ?MQTT_INTERNAL_EXTRA,
|
|
|
|
|
+ emqx_packet:info(properties, PktVar),
|
|
|
|
|
+ #{}
|
|
|
|
|
+ )
|
|
|
|
|
+ );
|
|
|
get_ctx(_) ->
|
|
get_ctx(_) ->
|
|
|
undefined.
|
|
undefined.
|
|
|
|
|
|
|
|
-from_extra(#{?EMQX_OTEL_CTX := OtelCtx}) ->
|
|
|
|
|
- OtelCtx;
|
|
|
|
|
-from_extra(_) ->
|
|
|
|
|
- undefined.
|
|
|
|
|
|
|
+from_extra(Key, Map) ->
|
|
|
|
|
+ maps:get(Key, Map, undefined).
|
|
|
|
|
|
|
|
-to_properties(OtelCtx, Props) ->
|
|
|
|
|
|
|
+to_properties(Key, OtelCtx, Props) ->
|
|
|
Extra = maps:get(?MQTT_INTERNAL_EXTRA, Props, #{}),
|
|
Extra = maps:get(?MQTT_INTERNAL_EXTRA, Props, #{}),
|
|
|
- Props#{?MQTT_INTERNAL_EXTRA => Extra#{?EMQX_OTEL_CTX => OtelCtx}}.
|
|
|
|
|
|
|
+ Props#{?MQTT_INTERNAL_EXTRA => Extra#{Key => OtelCtx}}.
|
|
|
|
|
+
|
|
|
|
|
+put_attrs(
|
|
|
|
|
+ Attrs,
|
|
|
|
|
+ #mqtt_packet{
|
|
|
|
|
+ variable = #mqtt_packet_publish{properties = Props} = PubPacket
|
|
|
|
|
+ } = Packet
|
|
|
|
|
+) ->
|
|
|
|
|
+ NProps = to_properties(?EMQX_OTEL_ATTRS, Attrs, Props),
|
|
|
|
|
+ Packet#mqtt_packet{variable = PubPacket#mqtt_packet_publish{properties = NProps}};
|
|
|
|
|
+put_attrs(
|
|
|
|
|
+ Attrs,
|
|
|
|
|
+ #mqtt_packet{
|
|
|
|
|
+ variable = #mqtt_packet_puback{properties = Props} = PubAckPacket
|
|
|
|
|
+ } = Packet
|
|
|
|
|
+) ->
|
|
|
|
|
+ NProps = to_properties(?EMQX_OTEL_ATTRS, Attrs, Props),
|
|
|
|
|
+ Packet#mqtt_packet{variable = PubAckPacket#mqtt_packet_puback{properties = NProps}};
|
|
|
|
|
+put_attrs(_Attrs, Any) ->
|
|
|
|
|
+ Any.
|
|
|
|
|
+
|
|
|
|
|
+get_attrs(#mqtt_packet{
|
|
|
|
|
+ variable = PktVar
|
|
|
|
|
+}) when is_tuple(PktVar) ->
|
|
|
|
|
+ from_extra(
|
|
|
|
|
+ ?EMQX_OTEL_ATTRS,
|
|
|
|
|
+ maps:get(
|
|
|
|
|
+ ?MQTT_INTERNAL_EXTRA,
|
|
|
|
|
+ emqx_packet:info(properties, PktVar),
|
|
|
|
|
+ #{}
|
|
|
|
|
+ )
|
|
|
|
|
+ ).
|
|
|
|
|
|
|
|
clear() ->
|
|
clear() ->
|
|
|
otel_ctx:clear().
|
|
otel_ctx:clear().
|