Rust MQTT telemetry - Githubissues

What's new

Trace MQTT message reception
Trace MQTT message publishing
Set the W3C context as MQTTv5 user property in published message
Extract W3C context from received message if any

Closes #124

How to test

Start a Jaeger/OTLP collector image

docker run --rm --name jaeger \
-e COLLECTOR_ZIPKIN_HOST_PORT=:9411 \
-p 6831:6831/udp \
-p 6832:6832/udp \
-p 5778:5778 \
-p 16686:16686 \
-p 4317:4317 \
-p 4318:4318 \
-p 14250:14250 \
-p 14268:14268 \
-p 14269:14269 \
-p 9411:9411 -d\
jaegertracing/all-in-one:1.58

Create the test script

wget https://github.com/user-attachments/files/17507082/test_copycat.txt &&
 mv test_copycat.txt /tmp/test_copycat.sh &&
 chmod +x /tmp/test_copycat.sh

(or download it as txt test_copycat.txt)

Listen to MQTT messages with MQTTv5 properties

mosquitto_sub -h test.mosquitto.org -p 1884 -u rw -P readwrite -v -t "default/+/v2x/cam/#" -V mqttv5 -F %j | jq -C

Edit the examples/config.ini file to fill the telemetry section with the following content
```
[telemetry]
host=localhost
port=4318
```

Start the copycat example with the telemetry feature

cargo run --example copycat --features geo_routing --features telemetry

Launch the test script
```
/tmp/test_copycat.sh
```
=> Several reception must be logged

DEBUG [copycat] item received: Packet { topic: GeoTopic { prefix: "default", queue: Out, suffix: "v2x", message_type: CAM, uuid: "ora_car_testcon2", geo_extension: Quadkey { tiles: [Two, Two, Two, Two, Two, Two, Two, Two, Two, Two, Two, Two, Two, Two, Two, Two, Two, Two, Two] } }, payload: Exchange { type_field: "cam", origin: "self", version: "1.2.0", source_uuid: "ora_car_2222", timestamp: 1618591358923, path: [], message: CAM(CooperativeAwarenessMessage { protocol_version: 2, station_id: 1077952580, generation_delta_time: 31571, basic_container: BasicContainer { station_type: Some(5), reference_position: ReferencePosition { latitude: 486230934, longitude: 22419064, altitude: 10000 }, confidence: Some(PositionConfidence { position_confidence_ellipse: Some(PositionConfidenceEllipse { semi_major_confidence: Some(0), semi_minor_confidence: Some(0), semi_major_orientation: Some(0) }), altitude: Some(1) }) }, high_frequency_container: HighFrequencyContainer { heading: Some(2375), speed: Some(137), drive_direction: Some(0), vehicle_length: Some(47), vehicle_width: Some(24), curvature: Some(0), curvature_calculation_mode: None, longitudinal_acceleration: Some(1), yaw_rate: Some(-27), acceleration_control: None, lane_position: Some(0), lateral_acceleration: Some(0), vertical_acceleration: Some(0), confidence: Some(HighFrequencyConfidence { heading: Some(2), speed: Some(3), vehicle_length: Some(1), yaw_rate: Some(6), longitudinal_acceleration: Some(5), curvature: Some(7), lateral_acceleration: None, vertical_acceleration: None }) }, low_frequency_container: Some(LowFrequencyContainer { vehicle_role: Some(0), exterior_lights: "00000000", path_history: [PathHistory { path_position: PathPosition { delta_latitude: None, delta_longitude: None, delta_altitude: None }, path_delta_time: Some(1) }] }) }) }, properties: PublishProperties { payload_format_indicator: None, message_expiry_interval: None, topic_alias: None, response_topic: None, correlation_data: None, user_properties: [], subscription_identifiers: [], content_type: None } }

=> One message publication must be logged

DEBUG [libits::client::application::pipeline] Packet to publish...
INFO [libits::client::application::pipeline] Cannot trace exchange, missing gateway component name in node configuration
DEBUG [libits::client::application::pipeline] Packet published!
DEBUG [rumqttc::v5::state] Publish. Topic = default/inQueue/v2x/cam/com_orange_its-client_10000/2/2/2/2/2/2/2/2/2/2/2/2/2/ 2/2/2/2/2/2, Pkid = 2, Payload Size = 962

The published message should be carrying the W3C context (logged by mosquitto_sub command)
```
{
 "tst": "2024-10-24T14:44:41.462008Z+0200",
 "topic": "default/inQueue/v2x/cam/com_orange_its-client_10000/2/2/2/2/2/2/2/2/2/2/2/2/2/2/2/2/2/2/2",
 "qos": 0,
 "retain": 0,
 "payloadlen": 962,
 "properties": {
   "user-properties": {
     "traceparent": "00-083c33ecb76bdd0f2d4b4b60e167f878-d5bb5974d8e3116b-01",
     "tracestate": ""
   }
 },
 "payload": "{\"type\":\"cam\",\"origin\":\"mec_application\",\"version\":\"1.2.0\",\"source_uuid\":\"com_orange_its-client_10000\",\"timestamp\":1729773881425,\"message\": {\"protocol_version\":2,\"station_id\":1077952580,\"generation_delta_time\":31571,\"basic_container\":{\"station_type\":5,\"reference_position\":{\"latitude\":486230934, \"longitude\":22419064,\"altitude\":10000},\"confidence\":{\"position_confidence_ellipse\":{\"semi_major_confidence\":0,\"semi_minor_confidence\":0, \"semi_major_orientation\":0},\"altitude\":1}},\"high_frequency_container\":{\"heading\":2375,\"speed\":137,\"drive_direction\":0,\"vehicle_length\":47,\"vehicle_width\":24, \"curvature\":0,\"longitudinal_acceleration\":1,\"yaw_rate\":-27,\"lane_position\":0,\"lateral_acceleration\":0,\"vertical_acceleration\":0,\"confidence\":{\"heading\":2, \"speed\":3,\"vehicle_length\":1,\"yaw_rate\":6,\"longitudinal_acceleration\":5,\"curvature\":7}},\"low_frequency_container\":{\"vehicle_role\":0, \"exterior_lights\":\"00000000\",\"path_history\":[{\"path_position\":{},\"path_delta_time\":1}]}}}"
}
```
The test scripts did prompt asking for a trace parent, answer it using the published message's one
Check the spans content on the local jaeger interface => You must see 8 traces
=> The 7th (the second most recent) must have a link to another span

Orange-OpenSource / its-client

Rust MQTT telemetry #182

What's new

How to test