feat(mqtt): support mqtt v5 for otel tracing

include/neuron/connection/mqtt_client.h

@@ -53,6 +53,11 @@ typedef enum {
 
 typedef struct neu_mqtt_client_s neu_mqtt_client_t;
 
+typedef struct {
+    char *traceparent;
+    char *tracestate;
+} trace_w3c_t;
+
 /**
  * Check that `topic_filter` is a valid MQTT topic filter.
  */
@@ -74,7 +79,8 @@ typedef void (*neu_mqtt_client_publish_cb_t)(int errcode, neu_mqtt_qos_e qos,
 typedef void (*neu_mqtt_client_subscribe_cb_t)(neu_mqtt_qos_e qos,
                                                const char *   topic,
                                                const uint8_t *payload,
-                                               uint32_t len, void *data);
+                                               uint32_t len, void *data,
+                                               trace_w3c_t *trace_w3c);
 
 neu_mqtt_client_t *neu_mqtt_client_new(neu_mqtt_version_e version);
 neu_mqtt_client_t *neu_mqtt_client_from_addr(const char *host, uint16_t port,
@@ -141,6 +147,9 @@ int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
 
 int neu_mqtt_client_unsubscribe(neu_mqtt_client_t *client, const char *topic);
 
+bool neu_mqtt_client_check_version_change(neu_mqtt_client_t *client,
+                                          neu_mqtt_version_e version);
+
 #ifdef __cplusplus
 }
 #endif

include/neuron/json/neu_json_mqtt.h

@@ -33,6 +33,8 @@ extern "C" {
 
 typedef struct {
     char *uuid;
+    char *traceparent;
+    char *tracestate;
 } neu_json_mqtt_t;
 
 typedef struct {

plugins/mqtt/aws-iot.json

@@ -1,4 +1,23 @@
 {
+  "version": {
+    "name": "MQTT Version",
+    "name_zh": "MQTT 版本",
+    "type": "map",
+    "attribute": "optional",
+    "default": 5,
+    "valid": {
+      "map": [
+        {
+          "key": "3.1.1",
+          "value": 4
+        },
+        {
+          "key": "5.0",
+          "value": 5
+        }
+      ]
+    }
+  },
   "client-id": {
     "name": "Client ID",
     "name_zh": "客户端 ID",

plugins/mqtt/azure-iot.json

@@ -1,4 +1,23 @@
 {
+  "version": {
+    "name": "MQTT Version",
+    "name_zh": "MQTT 版本",
+    "type": "map",
+    "attribute": "optional",
+    "default": 5,
+    "valid": {
+      "map": [
+        {
+          "key": "3.1.1",
+          "value": 4
+        },
+        {
+          "key": "5.0",
+          "value": 5
+        }
+      ]
+    }
+  },
   "device-id": {
     "name": "Device ID",
     "name_zh": "设备 ID",

plugins/mqtt/azure_iot_plugin.c

@@ -73,6 +73,12 @@ static int azure_parse_config(neu_plugin_t *plugin, const char *setting,
     char *      write_resp_topic = NULL;
     char *      username         = NULL;
 
+    neu_json_elem_t version = {
+        .name      = "version",
+        .t         = NEU_JSON_INT,
+        .v.val_int = NEU_MQTT_VERSION_V5,         // default to V5
+        .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
+    };
     neu_json_elem_t client_id = { .name = "device-id", .t = NEU_JSON_STR };
     neu_json_elem_t qos       = { .name = "qos", .t = NEU_JSON_INT };
     neu_json_elem_t format    = { .name = "format", .t = NEU_JSON_INT };
@@ -95,6 +101,14 @@ static int azure_parse_config(neu_plugin_t *plugin, const char *setting,
         goto error;
     }
 
+    ret = neu_parse_param(setting, &err_param, 1, &version);
+    if (0 != ret) {
+        plog_error(
+            plugin,
+            "parsing mqtt version fail, key: `%s`. Set default version: mqttv5",
+            err_param);
+    }
+
     // client-id, required
     if (0 == strlen(client_id.v.val_str)) {
         plog_error(plugin, "setting empty client-id");
@@ -165,6 +179,7 @@ static int azure_parse_config(neu_plugin_t *plugin, const char *setting,
         goto error;
     }
 
+    config->version             = version.v.val_int;
     config->client_id           = client_id.v.val_str;
     config->qos                 = qos.v.val_int;
     config->format              = format.v.val_int;

plugins/mqtt/mqtt.json

@@ -1,4 +1,23 @@
 {
+  "version": {
+    "name": "MQTT Version",
+    "name_zh": "MQTT 版本",
+    "type": "map",
+    "attribute": "optional",
+    "default": 5,
+    "valid": {
+      "map": [
+        {
+          "key": "3.1.1",
+          "value": 4
+        },
+        {
+          "key": "5.0",
+          "value": 5
+        }
+      ]
+    }
+  },
   "client-id": {
     "name": "Client ID",
     "name_zh": "客户端 ID",

plugins/mqtt/mqtt_config.c

@@ -218,6 +218,12 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
     char *      err_param   = NULL;
     const char *placeholder = "********";
 
+    neu_json_elem_t version = {
+        .name      = "version",
+        .t         = NEU_JSON_INT,
+        .v.val_int = NEU_MQTT_VERSION_V5,         // default to V5
+        .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, // for backward compatibility
+    };
     neu_json_elem_t client_id = { .name = "client-id", .t = NEU_JSON_STR };
     neu_json_elem_t qos       = {
         .name      = "qos",
@@ -268,6 +274,14 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
         goto error;
     }
 
+    ret = neu_parse_param(setting, &err_param, 1, &version);
+    if (0 != ret) {
+        plog_error(
+            plugin,
+            "parsing mqtt version fail, key: `%s`. Set default version: mqttv5",
+            err_param);
+    }
+
     // client-id, required
     if (0 == strlen(client_id.v.val_str)) {
         plog_error(plugin, "setting empty client-id");
@@ -340,6 +354,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
         goto error;
     }
 
+    config->version             = version.v.val_int;
     config->client_id           = client_id.v.val_str;
     config->qos                 = qos.v.val_int;
     config->format              = format.v.val_int;
@@ -359,6 +374,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
     config->key                 = key.v.val_str;
     config->keypass             = keypass.v.val_str;
 
+    plog_notice(plugin, "config MQTT version    : %d", config->version);
     plog_notice(plugin, "config client-id       : %s", config->client_id);
     plog_notice(plugin, "config qos             : %d", config->qos);
     plog_notice(plugin, "config format          : %s",

plugins/mqtt/mqtt_config.h

@@ -48,6 +48,7 @@ static inline const char *mqtt_upload_format_str(mqtt_upload_format_e f)
 }
 
 typedef struct {
+    neu_mqtt_version_e   version;             // mqtt version
     char *               client_id;           // client id
     neu_mqtt_qos_e       qos;                 // message QoS
     mqtt_upload_format_e format;              // upload format

plugins/mqtt/mqtt_handle.c

@@ -133,6 +133,11 @@ static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
     plog_notice(plugin, "read uuid:%s, group:%s, node:%s", mqtt->uuid,
                 req->group, req->node);
 
+    if (mqtt->traceparent && mqtt->tracestate) {
+        plog_notice(plugin, "read, traceparent:%s, tracestate:%s",
+                    mqtt->traceparent, mqtt->tracestate);
+    }
+
     neu_reqresp_head_t header = { 0 };
     header.ctx                = mqtt;
     header.type               = NEU_REQ_READ_GROUP;
@@ -186,9 +191,14 @@ static int json_value_to_tag_value(union neu_json_value *req,
 static int send_write_tag_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
                               neu_json_write_req_t *req)
 {
-    plog_notice(plugin, "write tag uuid:%s, group:%s, node:%s", mqtt->uuid,
+    plog_notice(plugin, "write tag, uuid:%s, group:%s, node:%s", mqtt->uuid,
                 req->group, req->node);
 
+    if (mqtt->traceparent && mqtt->tracestate) {
+        plog_notice(plugin, "write tag, traceparent:%s, tracestate:%s",
+                    mqtt->traceparent, mqtt->tracestate);
+    }
+
     neu_reqresp_head_t  header = { 0 };
     neu_req_write_tag_t cmd    = { 0 };
 
@@ -221,6 +231,11 @@ static int send_write_tags_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
     plog_notice(plugin, "write tags uuid:%s, group:%s, node:%s", mqtt->uuid,
                 req->group, req->node);
 
+    if (mqtt->traceparent && mqtt->tracestate) {
+        plog_notice(plugin, "write tag, traceparent:%s, tracestate:%s",
+                    mqtt->traceparent, mqtt->tracestate);
+    }
+
     for (int i = 0; i < req->n_tag; i++) {
         if (req->tags[i].t == NEU_JSON_STR) {
             if (strlen(req->tags[i].value.val_str) >= NEU_VALUE_SIZE) {
@@ -307,7 +322,8 @@ int publish(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic,
 }
 
 void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
-                      const uint8_t *payload, uint32_t len, void *data)
+                      const uint8_t *payload, uint32_t len, void *data,
+                      trace_w3c_t *trace_w3c)
 {
     int               rv     = 0;
     neu_plugin_t *    plugin = data;
@@ -340,6 +356,13 @@ void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
         return;
     }
 
+    if (trace_w3c && trace_w3c->traceparent) {
+        mqtt->traceparent = strdup(trace_w3c->traceparent);
+    }
+    if (trace_w3c && trace_w3c->tracestate) {
+        mqtt->tracestate = strdup(trace_w3c->tracestate);
+    }
+
     rv = neu_json_decode_write(json_str, &req);
     if (0 != rv) {
         plog_error(plugin, "neu_json_decode_write fail");
@@ -398,7 +421,8 @@ end:
 }
 
 void handle_read_req(neu_mqtt_qos_e qos, const char *topic,
-                     const uint8_t *payload, uint32_t len, void *data)
+                     const uint8_t *payload, uint32_t len, void *data,
+                     trace_w3c_t *trace_w3c)
 {
     int                  rv     = 0;
     neu_plugin_t *       plugin = data;
@@ -431,6 +455,13 @@ void handle_read_req(neu_mqtt_qos_e qos, const char *topic,
         return;
     }
 
+    if (trace_w3c && trace_w3c->traceparent) {
+        mqtt->traceparent = strdup(trace_w3c->traceparent);
+    }
+    if (trace_w3c && trace_w3c->tracestate) {
+        mqtt->tracestate = strdup(trace_w3c->tracestate);
+    }
+
     rv = neu_json_decode_read_req(json_str, &req);
     if (0 != rv) {
         plog_error(plugin, "neu_json_decode_read_req fail");

plugins/mqtt/mqtt_handle.h

@@ -32,13 +32,15 @@ int publish(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic,
             char *payload, size_t payload_len);
 
 void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
-                      const uint8_t *payload, uint32_t len, void *data);
+                      const uint8_t *payload, uint32_t len, void *data,
+                      trace_w3c_t *trace_w3c);
 
 int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
                           neu_resp_error_t *data);
 
 void handle_read_req(neu_mqtt_qos_e qos, const char *topic,
-                     const uint8_t *payload, uint32_t len, void *data);
+                     const uint8_t *payload, uint32_t len, void *data,
+                     trace_w3c_t *trace_w3c);
 
 int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
                          neu_resp_read_group_t *data);

plugins/mqtt/mqtt_plugin_intf.c

@@ -260,7 +260,7 @@ int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
     }
 
     if (NULL == plugin->client) {
-        plugin->client = neu_mqtt_client_new(NEU_MQTT_VERSION_V311);
+        plugin->client = neu_mqtt_client_new(config.version);
         if (NULL == plugin->client) {
             plog_error(plugin, "neu_mqtt_client_new fail");
             rv = NEU_ERR_EINTERNAL;
@@ -275,6 +275,16 @@ int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
             rv = NEU_ERR_EINTERNAL;
             goto error;
         }
+        if (neu_mqtt_client_check_version_change(plugin->client,
+                                                 config.version)) {
+            neu_mqtt_client_free(plugin->client);
+            plugin->client = neu_mqtt_client_new(config.version);
+        }
+    } else if (neu_mqtt_client_check_version_change(plugin->client,
+                                                    config.version)) {
+        // plugin stopped and version changed
+        neu_mqtt_client_free(plugin->client);
+        plugin->client = neu_mqtt_client_new(config.version);
     }
 
     rv = config_mqtt_client(plugin, plugin->client, &config);

src/connection/mqtt_client.c

@@ -427,6 +427,45 @@ end:
     return;
 }
 
+void neu_mqtt_property_foreach(property *prop,
+                               void (*cb)(property *, trace_w3c_t *),
+                               trace_w3c_t *trace_w3c)
+{
+    for (property *p = prop->next; p != NULL; p = p->next) {
+        cb(p, trace_w3c);
+    }
+}
+
+void neu_mqtt_decode_property_trace(property *prop, trace_w3c_t *trace_w3c)
+{
+    if (prop == NULL) {
+        return;
+    }
+
+    uint8_t type = prop->data.p_type;
+
+    switch (type) {
+    case STR_PAIR:
+        if (strncmp((const char *) prop->data.p_value.strpair.key.buf,
+                    "traceparent",
+                    prop->data.p_value.strpair.key.length) == 0) {
+            trace_w3c->traceparent =
+                strndup((const char *) prop->data.p_value.strpair.value.buf,
+                        prop->data.p_value.strpair.value.length);
+        } else if (strncmp((const char *) prop->data.p_value.strpair.key.buf,
+                           "tracestate",
+                           prop->data.p_value.strpair.key.length) == 0) {
+            trace_w3c->tracestate =
+                strndup((const char *) prop->data.p_value.strpair.value.buf,
+                        prop->data.p_value.strpair.value.length);
+        }
+        break;
+
+    default:
+        break;
+    }
+}
+
 static void task_handle_recv(task_t *task, neu_mqtt_client_t *client)
 {
     (void) client;
@@ -435,10 +474,28 @@ static void task_handle_recv(task_t *task, neu_mqtt_client_t *client)
     uint32_t payload_len;
     uint8_t *payload = nng_mqtt_msg_get_publish_payload(msg, &payload_len);
 
+    trace_w3c_t trace_w3c = { NULL, NULL };
+
+    if (client->version == MQTT_PROTOCOL_VERSION_v5) {
+        property *plist = nng_mqtt_msg_get_publish_property(msg);
+        if (plist != NULL) {
+            neu_mqtt_property_foreach(plist, neu_mqtt_decode_property_trace,
+                                      &trace_w3c);
+        }
+    }
+
     task->recv.sub->cb(task->recv.sub->qos, task->recv.sub->topic, payload,
-                       payload_len, task->recv.sub->data);
+                       payload_len, task->recv.sub->data, &trace_w3c);
 
     nng_msg_free(msg);
+    if (trace_w3c.traceparent) {
+        free(trace_w3c.traceparent);
+        trace_w3c.traceparent = NULL;
+    }
+    if (trace_w3c.tracestate) {
+        free(trace_w3c.tracestate);
+        trace_w3c.tracestate = NULL;
+    }
 }
 
 static subscription_t *subscription_new(neu_mqtt_client_t *client,
@@ -757,6 +814,23 @@ static int client_send_sub_msg(neu_mqtt_client_t *client,
     nng_mqtt_msg_set_packet_type(sub_msg, NNG_MQTT_SUBSCRIBE);
     nng_mqtt_msg_set_subscribe_topics(sub_msg, topic_qos, topic_qos_count);
 
+    /*if (client->version == MQTT_PROTOCOL_VERSION_v5) {
+        property *plist = mqtt_property_alloc();
+        property *p1 = mqtt_property_set_value_strpair(USER_PROPERTY,
+    "traceparent", strlen("traceparent"), "abcde", strlen("abcde"), true);
+            mqtt_property_append(plist, p1);
+        property *p2 = mqtt_property_set_value_strpair(USER_PROPERTY,
+    "tracestate", strlen("tracestate"), "abcde", strlen("abcde"), true);
+            mqtt_property_append(plist, p2);
+
+        nng_mqtt_msg_set_subscribe_property(sub_msg, plist);
+    }*/
+
+    /*nng_mqttv5_msg_encode(sub_msg);
+    uint8_t print[1024] = {0};
+    nng_mqtt_msg_dump(sub_msg, print, 1024, true);
+        printf("%s\n", print);*/
+
     task_t *task = client_alloc_task(client);
     if (NULL == task) {
         log(error, "client_alloc_task fail");
@@ -1367,15 +1441,20 @@ int neu_mqtt_client_open(neu_mqtt_client_t *client)
         return 0;
     }
 
-    if (NEU_MQTT_VERSION_V5 == client->version &&
-        (rv = nng_mqttv5_client_open(&client->sock)) != 0) {
-        log(error, "nng_mqttv5_client_open fail: %s", nng_strerror(rv));
-        nng_mtx_unlock(client->mtx);
-        return -1;
-    } else if ((rv = nng_mqtt_client_open(&client->sock)) != 0) {
-        log(error, "nng_mqtt_client_open fail: %s", nng_strerror(rv));
-        nng_mtx_unlock(client->mtx);
-        return -1;
+    if (NEU_MQTT_VERSION_V5 == client->version) {
+        rv = nng_mqttv5_client_open(&client->sock);
+        if (rv != 0) {
+            log(error, "nng_mqttv5_client_open fail: %s", nng_strerror(rv));
+            nng_mtx_unlock(client->mtx);
+            return -1;
+        }
+    } else {
+        rv = nng_mqtt_client_open(&client->sock);
+        if (rv != 0) {
+            log(error, "nng_mqtt_client_open fail: %s", nng_strerror(rv));
+            nng_mtx_unlock(client->mtx);
+            return -1;
+        }
     }
 
     if (0 != client_make_url(client)) {
@@ -1516,6 +1595,18 @@ int neu_mqtt_client_publish(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
     nng_mqtt_msg_set_publish_payload(pub_msg, (uint8_t *) payload, len);
     nng_mqtt_msg_set_publish_qos(pub_msg, qos);
 
+    /*if (client->version == MQTT_PROTOCOL_VERSION_v5) {
+        property *plist = mqtt_property_alloc();
+        property *p1 = mqtt_property_set_value_strpair(USER_PROPERTY,
+    "traceparent", strlen("traceparent"), "abcde", strlen("abcde"), true);
+            mqtt_property_append(plist, p1);
+        property *p2 = mqtt_property_set_value_strpair(USER_PROPERTY,
+    "tracestate", strlen("tracestate"), "12345", strlen("12345"), true);
+            mqtt_property_append(plist, p2);
+
+        nng_mqtt_msg_set_publish_property(pub_msg, plist);
+    }*/
+
     nng_mtx_lock(client->mtx);
     task = client_alloc_task(client);
     nng_mtx_unlock(client->mtx);
@@ -1626,3 +1717,12 @@ error:
     nng_mtx_unlock(client->mtx);
     return -1;
 }
+
+bool neu_mqtt_client_check_version_change(neu_mqtt_client_t *client,
+                                          neu_mqtt_version_e version)
+{
+    if (client != NULL) {
+        return client->version != version;
+    }
+    return true;
+}

src/parser/neu_json_mqtt.c

@@ -66,6 +66,12 @@ void neu_json_decode_mqtt_req_free(neu_json_mqtt_t *req)
     if (NULL != req->uuid) {
         free(req->uuid);
     }
+    if (NULL != req->traceparent) {
+        free(req->traceparent);
+    }
+    if (NULL != req->tracestate) {
+        free(req->tracestate);
+    }
 
     free(req);
 }