Passing the traceID over a message bus that is NOT MQTT

[proffalken]

Hello,

I'm working on adding tracing to my ROS2 application.

ROS has it's own internal message bus system and I can exchange JSON strings, but it is NOT MQTT, so I can't use any of the MQTT context propagation code.

At the moment, the PRODUCER is as follows:

    def timer_callback(self):
        msg = String()
        main_angle = random.randrange(0, 45)
        with self.tracer.start_as_current_span("publish_message", SpanKind="PRODUCER"):
            current_span = trace.get_current_span()
            payload = {
                    "servos": [
                        {"angle": 95},
                        {"angle": main_angle},
                        {"angle": main_angle},
                        ],
                        "span_ctx": current_span.get_span_context().trace_id
                    }
            msg.data = json.dumps(payload)
            self.publisher_.publish(msg)
            self.otel_log.info('Publishing: "%s"' % msg.data)
            self.get_logger().info('Publishing: "%s"' % msg.data)
            self.i += 1

This results in the following message being sent:

{"servos": [{"angle": 95}, {"angle": 27}, {"angle": 27}], "span_ctx": 284478121259741876389388285130937154007}

What I then want to do is extract the trace ID (span_ctx in my JSON message) and set that as the parent of the span in the subscriber/consumer:

    def listener_callback(self, msg):
        payload = json.loads(msg.data)
        ######## payload['span_ctx'] holds the TraceID #####
        with self.tracer.start_as_current_span("recv_message", SpanKind="CONSUMER"):
            self.otel_log.info('I heard: "%s"' % msg.data)
            self.get_logger().info('I heard: "%s"' % msg.data)
            s = 0
            while s < len(payload['servos']):
                    diff = self.kit.servo[s].angle = payload['servos'][s]['angle']
                    # If the value is less than 90, work our way up, 
                    # otherwise work our way down
                    if diff < 90 and diff != 0:
                        self.otel_log.info(f"Difference is {diff}")
                        c = self.kit.servo[s].angle
                        self.kit.servo[s].angle = c + 1
                        self.otel_log.info(f"Delay set to {5/diff}")
                        sleep(5/diff)
                    elif diff > 90:
                        c = self.kit.servo[s].angle
                        self.kit.servo[s].angle = c - 1
                        self.otel_log.info(f"Delay set to {diff/5}")
                        sleep(5/diff)
                    else:
                        self.otel_log.info("No difference found")
                    s = s+1

Is there a way to set the parent span or traceID manually without passing the entire context?

[malcolmrebughini]

You can use the propagate package:

The producer could look something like this:

    def timer_callback(self):
        msg = String()
        main_angle = random.randrange(0, 45)
        with self.tracer.start_as_current_span("publish_message", SpanKind="PRODUCER"):
            payload = {
                    "servos": [
                        {"angle": 95},
                        {"angle": main_angle},
                        {"angle": main_angle},
                        ],
                    }
            propagate.get_global_textmap().inject(payload) # New line
            msg.data = json.dumps(payload)
            self.publisher_.publish(msg)
            self.otel_log.info('Publishing: "%s"' % msg.data)
            self.get_logger().info('Publishing: "%s"' % msg.data)
            self.i += 1

The payload will get a couple of new keys:

{'servos': 'something', 'traceparent': '00-b54fb9a54271c1fde422d197f7dc08eb-a8c11dfd4a7f3e31-01', 'baggage': 'key=value'}

And the consumer:

from opentelemetry import context

   def listener_callback(self, msg):
        payload = json.loads(msg.data)
        ctx = propagate.get_global_textmap().extract(payload)
        context.attach(ctx)

        with self.tracer.start_as_current_span("recv_message", SpanKind="CONSUMER"):
            self.otel_log.info('I heard: "%s"' % msg.data)
            self.get_logger().info('I heard: "%s"' % msg.data)
            s = 0
            while s < len(payload['servos']):
                    diff = self.kit.servo[s].angle = payload['servos'][s]['angle']
                    # If the value is less than 90, work our way up, 
                    # otherwise work our way down
                    if diff < 90 and diff != 0:
                        self.otel_log.info(f"Difference is {diff}")
                        c = self.kit.servo[s].angle
                        self.kit.servo[s].angle = c + 1
                        self.otel_log.info(f"Delay set to {5/diff}")
                        sleep(5/diff)
                    elif diff > 90:
                        c = self.kit.servo[s].angle
                        self.kit.servo[s].angle = c - 1
                        self.otel_log.info(f"Delay set to {diff/5}")
                        sleep(5/diff)
                    else:
                        self.otel_log.info("No difference found")
                    s = s+1

Hope this helps

[proffalken]

This works beautifully, I can now visualise the relationships between ROS2 nodes in Grafana, thanks!