From 4981173ceed1a2e71a7771f4c2ed3d800d68e4fb Mon Sep 17 00:00:00 2001 From: hxy7yx <1595670487@qq.com> Date: Wed, 4 Sep 2024 16:10:29 +0800 Subject: [PATCH] feat(mqtt): support mqtt v5 for otel tracing --- include/neuron/connection/mqtt_client.h | 11 ++- include/neuron/json/neu_json_mqtt.h | 2 + plugins/mqtt/aws-iot.json | 19 ++++ plugins/mqtt/azure-iot.json | 19 ++++ plugins/mqtt/azure_iot_plugin.c | 15 +++ plugins/mqtt/mqtt.json | 19 ++++ plugins/mqtt/mqtt_config.c | 16 ++++ plugins/mqtt/mqtt_config.h | 1 + plugins/mqtt/mqtt_handle.c | 37 +++++++- plugins/mqtt/mqtt_handle.h | 6 +- plugins/mqtt/mqtt_plugin_intf.c | 12 ++- src/connection/mqtt_client.c | 120 ++++++++++++++++++++++-- src/parser/neu_json_mqtt.c | 6 ++ 13 files changed, 266 insertions(+), 17 deletions(-) diff --git a/include/neuron/connection/mqtt_client.h b/include/neuron/connection/mqtt_client.h index 135de534..a1f5809a 100644 --- a/include/neuron/connection/mqtt_client.h +++ b/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 diff --git a/include/neuron/json/neu_json_mqtt.h b/include/neuron/json/neu_json_mqtt.h index 9c7cc292..fb850b51 100644 --- a/include/neuron/json/neu_json_mqtt.h +++ b/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 { diff --git a/plugins/mqtt/aws-iot.json b/plugins/mqtt/aws-iot.json index 237e8065..8ad79c18 100644 --- a/plugins/mqtt/aws-iot.json +++ b/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", diff --git a/plugins/mqtt/azure-iot.json b/plugins/mqtt/azure-iot.json index 14fd6ecd..aca21359 100644 --- a/plugins/mqtt/azure-iot.json +++ b/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", diff --git a/plugins/mqtt/azure_iot_plugin.c b/plugins/mqtt/azure_iot_plugin.c index 82ba6147..5fe05db4 100644 --- a/plugins/mqtt/azure_iot_plugin.c +++ b/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; diff --git a/plugins/mqtt/mqtt.json b/plugins/mqtt/mqtt.json index 493c0172..7862d523 100644 --- a/plugins/mqtt/mqtt.json +++ b/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", diff --git a/plugins/mqtt/mqtt_config.c b/plugins/mqtt/mqtt_config.c index 0be384a6..6a9ced31 100644 --- a/plugins/mqtt/mqtt_config.c +++ b/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", diff --git a/plugins/mqtt/mqtt_config.h b/plugins/mqtt/mqtt_config.h index e1cbccf1..969f2be4 100644 --- a/plugins/mqtt/mqtt_config.h +++ b/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 diff --git a/plugins/mqtt/mqtt_handle.c b/plugins/mqtt/mqtt_handle.c index 74cfcfc4..15bf06f0 100644 --- a/plugins/mqtt/mqtt_handle.c +++ b/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"); diff --git a/plugins/mqtt/mqtt_handle.h b/plugins/mqtt/mqtt_handle.h index dc62ef92..be3ba3b5 100644 --- a/plugins/mqtt/mqtt_handle.h +++ b/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); diff --git a/plugins/mqtt/mqtt_plugin_intf.c b/plugins/mqtt/mqtt_plugin_intf.c index 66ae96de..12ee0d22 100644 --- a/plugins/mqtt/mqtt_plugin_intf.c +++ b/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); diff --git a/src/connection/mqtt_client.c b/src/connection/mqtt_client.c index fb8619f4..1d94a914 100644 --- a/src/connection/mqtt_client.c +++ b/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; +} \ No newline at end of file diff --git a/src/parser/neu_json_mqtt.c b/src/parser/neu_json_mqtt.c index 5a8de061..fcd3563d 100644 --- a/src/parser/neu_json_mqtt.c +++ b/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); }