feat(mqtt): add support for context propagation for mqtt nodes if mqtt v5 is used (#6)

Co-authored-by: Andrej Krpic <[email protected]>

Author: akrpic77

lib/opentelemetry-node.js

@@ -145,8 +145,10 @@ function createSpan (tracer, msg, nodeDefinition, node, isNotTraced) {
       if (nodeDefinition.type === 'http in') {
         // try to get trace context in incoming http request headers
         ctx = propagator.extract(context.active(), msg.req.headers, defaultTextMapGetter)
-      }
-      if (nodeDefinition.type === 'amqp-in') {
+      } else if (nodeDefinition.type === 'mqtt in' && msg.userProperties) {
+        // try to get trace context in incoming mqtt v5 user properties
+        ctx = propagator.extract(context.active(), msg.userProperties, defaultTextMapGetter)
+      } else if (nodeDefinition.type === 'amqp-in') {
         // try to get trace context in incoming ampq message headers
         ctx = propagator.extract(context.active(), msg.properties.headers, defaultTextMapGetter)
       }
@@ -348,14 +350,25 @@ module.exports = function (RED) {
       logEvent(node, '4.postDeliver', sendEvent)
       const span = createSpan(tracer, sendEvent.msg, sendEvent.destination.node, node, ignoredTypesList.includes(sendEvent.destination.node.type))
       if (propagateHeadersTypesList.includes(sendEvent.destination.node.type)) {
-        // add trace context in http request headers
         const output = {}
         const ctx = trace.setSpan(context.active(), span)
         propagation.inject(ctx, output)
-        if (!sendEvent.msg.headers) {
-          sendEvent.msg.headers = {}
+        switch (sendEvent.destination.node.type) {
+          // add trace context in mqtt v5 user properties
+          case 'mqtt out':
+            if (!sendEvent.msg.userProperties) {
+              sendEvent.msg.userProperties = {}
+            }
+            Object.assign(sendEvent.msg.userProperties, output)
+            break
+          default:
+            // add trace context in http request headers
+            if (!sendEvent.msg.headers) {
+              sendEvent.msg.headers = {}
+            }
+            Object.assign(sendEvent.msg.headers, output)
+            break
         }
-        Object.assign(sendEvent.msg.headers, output)
       }
       if (sendEvent.source.node.type === 'switch' || sendEvent.source.node.type.startsWith('subflow')) {
         // end switch or subflow spans as they do not trigger onComplete