Allow usage of the new MQTT 5 implementation (#12838)
* adds support for new MQTT5 implementation in agent, currently by default disabled as Tech Preview
Timotej S committed
Jun 6, 2022 at 12:05 UTC
123e923f76ad2cc458dd98b01ba8b29f2306bee2
17 files changed
+145
-129
CMakeLists.txt
+5
@@ -817,6 +817,11 @@ set(ACLK_FILES
817
mqtt_websockets/src/include/mqtt_wss_log.h
818
mqtt_websockets/src/ws_client.c
819
mqtt_websockets/src/include/ws_client.h
820
+ mqtt_websockets/src/mqtt_ng.c
821
+ mqtt_websockets/src/include/mqtt_ng.h
822
+ mqtt_websockets/src/common_public.c
823
+ mqtt_websockets/src/include/common_public.h
824
+ mqtt_websockets/src/include/common_internal.h
825
mqtt_websockets/c-rbuf/src/ringbuffer.c
826
mqtt_websockets/c-rbuf/include/ringbuffer.h
827
mqtt_websockets/c-rbuf/src/ringbuffer_internal.h
Makefile.am
+5
@@ -628,6 +628,11 @@ ACLK_FILES = \
628
mqtt_websockets/src/include/mqtt_wss_log.h \
629
mqtt_websockets/src/ws_client.c \
630
mqtt_websockets/src/include/ws_client.h \
631
+ mqtt_websockets/src/mqtt_ng.c \
632
+ mqtt_websockets/src/include/mqtt_ng.h \
633
+ mqtt_websockets/src/common_public.c \
634
+ mqtt_websockets/src/include/common_public.h \
635
+ mqtt_websockets/src/include/common_internal.h \
636
mqtt_websockets/c-rbuf/src/ringbuffer.c \
637
mqtt_websockets/c-rbuf/include/ringbuffer.h \
638
mqtt_websockets/c-rbuf/src/ringbuffer_internal.h \
aclk/README.md
+2
@@ -50,10 +50,12 @@ You can configure following keys in the `netdata.conf` section `[cloud]`:
50
[cloud]
51
statistics = yes
52
query thread count = 2
53
+ mqtt5 = no
54
```
55
56
- `statistics` enables/disables ACLK related statistics and their charts. You can disable this to save some space in the database and slightly reduce memory usage of Netdata Agent.
57
- `query thread count` specifies the number of threads to process cloud queries. Increasing this setting is useful for nodes with many children (streaming), which can expect to handle more queries (and/or more complicated queries).
58
+- `mqtt5` enables the new MQTT5 protocol implementation in the Agent. Currently a technical preview.
59
60
## Disable the ACLK
61
aclk/aclk.c
+65
-33
@@ -12,6 +12,7 @@
12
#include "aclk_rx_msgs.h"
13
#include "aclk_collector_list.h"
14
#include "https_client.h"
15
+#include "schema-wrappers/schema_wrappers.h"
16
17
#include "aclk_proxy.h"
18
@@ -172,7 +173,7 @@ void aclk_mqtt_wss_log_cb(mqtt_wss_log_type_t log_type, const char* str)
173
case MQTT_WSS_LOG_ERROR:
174
case MQTT_WSS_LOG_FATAL:
175
case MQTT_WSS_LOG_WARN:
175
- error("%s", str);
176
+ error_report("%s", str);
177
return;
178
case MQTT_WSS_LOG_INFO:
179
info("%s", str);
@@ -391,7 +392,7 @@ static inline void queue_connect_payloads(void)
392
393
static inline void mqtt_connected_actions(mqtt_wss_client client)
394
{
394
- const char *topic = aclk_get_topic(ACLK_TOPICID_COMMAND);
395
+ char *topic = (char*)aclk_get_topic(ACLK_TOPICID_COMMAND);
396
397
if (!topic)
398
error("Unable to fetch topic for COMMAND (to subscribe)");
@@ -400,7 +401,7 @@ static inline void mqtt_connected_actions(mqtt_wss_client client)
401
402
#ifdef ENABLE_NEW_CLOUD_PROTOCOL
403
if (aclk_use_new_cloud_arch) {
403
- topic = aclk_get_topic(ACLK_TOPICID_CMD_NG_V1);
404
+ topic = (char*)aclk_get_topic(ACLK_TOPICID_CMD_NG_V1);
405
if (!topic)
406
error("Unable to fetch topic for protobuf COMMAND (to subscribe)");
407
else
@@ -800,10 +801,12 @@ void *aclk_main(void *ptr)
801
if (wait_till_agent_claim_ready())
802
goto exit;
803
804
+ use_mqtt_5 = config_get_boolean(CONFIG_SECTION_CLOUD, "mqtt5", CONFIG_BOOLEAN_NO);
805
+
806
#ifdef ENABLE_NEW_CLOUD_PROTOCOL
804
- if (!(mqttwss_client = mqtt_wss_new("mqtt_wss", aclk_mqtt_wss_log_cb, msg_callback, puback_callback))) {
807
+ if (!(mqttwss_client = mqtt_wss_new("mqtt_wss", aclk_mqtt_wss_log_cb, msg_callback, puback_callback, use_mqtt_5))) {
808
#else
806
- if (!(mqttwss_client = mqtt_wss_new("mqtt_wss", aclk_mqtt_wss_log_cb, msg_callback_old_protocol, puback_callback))) {
809
+ if (!(mqttwss_client = mqtt_wss_new("mqtt_wss", aclk_mqtt_wss_log_cb, msg_callback_old_protocol, puback_callback, use_mqtt_5))) {
810
#endif
811
error("Couldn't initialize MQTT_WSS network library");
812
goto exit;
@@ -1041,6 +1044,7 @@ void aclk_del_collector(RRDHOST *host, const char *plugin_name, const char *modu
1044
aclk_queue_query(query);
1045
}
1046
1047
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
1048
void aclk_host_state_update(RRDHOST *host, int cmd)
1049
{
1050
uuid_t node_id;
@@ -1060,28 +1064,40 @@ void aclk_host_state_update(RRDHOST *host, int cmd)
1064
aclk_query_t create_query;
1065
create_query = aclk_query_new(REGISTER_NODE);
1066
rrdhost_aclk_state_lock(localhost);
1063
- create_query->data.node_creation.claim_id = strdupz(localhost->aclk_state.claimed_id);
1067
+ node_instance_creation_t node_instance_creation = {
1068
+ .claim_id = localhost->aclk_state.claimed_id,
1069
+ .hops = host->system_info->hops,
1070
+ .hostname = host->hostname,
1071
+ .machine_guid = host->machine_guid
1072
+ };
1073
+ create_query->data.bin_payload.payload = generate_node_instance_creation(&create_query->data.bin_payload.size, &node_instance_creation);
1074
rrdhost_aclk_state_unlock(localhost);
1065
- create_query->data.node_creation.hops = (uint32_t) host->system_info->hops;
1066
- create_query->data.node_creation.hostname = strdupz(host->hostname);
1067
- create_query->data.node_creation.machine_guid = strdupz(host->machine_guid);
1075
+ create_query->data.bin_payload.topic = ACLK_TOPICID_CREATE_NODE;
1076
+ create_query->data.bin_payload.msg_name = "CreateNodeInstance";
1077
info("Registering host=%s, hops=%u",host->machine_guid, host->system_info->hops);
1078
aclk_queue_query(create_query);
1079
return;
1080
}
1081
1082
aclk_query_t query = aclk_query_new(NODE_STATE_UPDATE);
1074
- query->data.node_update.hops = (uint32_t) host->system_info->hops;
1083
+ node_instance_connection_t node_state_update = {
1084
+ .hops = host->system_info->hops,
1085
+ .live = cmd,
1086
+ .queryable = 1,
1087
+ .session_id = aclk_session_newarch
1088
+ };
1089
+ node_state_update.node_id = mallocz(UUID_STR_LEN);
1090
+ uuid_unparse_lower(node_id, (char*)node_state_update.node_id);
1091
rrdhost_aclk_state_lock(localhost);
1076
- query->data.node_update.claim_id = strdupz(localhost->aclk_state.claimed_id);
1092
+ node_state_update.claim_id = localhost->aclk_state.claimed_id;
1093
+ query->data.bin_payload.payload = generate_node_instance_connection(&query->data.bin_payload.size, &node_state_update);
1094
rrdhost_aclk_state_unlock(localhost);
1078
- query->data.node_update.live = cmd;
1079
- query->data.node_update.node_id = mallocz(UUID_STR_LEN);
1080
- uuid_unparse_lower(node_id, (char*)query->data.node_update.node_id);
1081
- query->data.node_update.queryable = 1;
1082
- query->data.node_update.session_id = aclk_session_newarch;
1083
- info("Queuing status update for node=%s, live=%d, hops=%u",(char*)query->data.node_update.node_id, cmd,
1095
+
1096
+ info("Queuing status update for node=%s, live=%d, hops=%u",(char*)node_state_update.node_id, cmd,
1097
host->system_info->hops);
1098
+ freez((void*)node_state_update.node_id);
1099
+ query->data.bin_payload.msg_name = "UpdateNodeInstanceConnection";
1100
+ query->data.bin_payload.topic = ACLK_TOPICID_NODE_CONN;
1101
aclk_queue_query(query);
1102
}
1103
@@ -1096,39 +1112,52 @@ void aclk_send_node_instances()
1112
while (!uuid_is_null(list->host_id)) {
1113
if (!uuid_is_null(list->node_id)) {
1114
aclk_query_t query = aclk_query_new(NODE_STATE_UPDATE);
1115
+ node_instance_connection_t node_state_update = {
1116
+ .live = list->live,
1117
+ .hops = list->hops,
1118
+ .queryable = 1,
1119
+ .session_id = aclk_session_newarch
1120
+ };
1121
+ node_state_update.node_id = mallocz(UUID_STR_LEN);
1122
+ uuid_unparse_lower(list->node_id, (char*)node_state_update.node_id);
1123
rrdhost_aclk_state_lock(localhost);
1100
- query->data.node_update.claim_id = strdupz(localhost->aclk_state.claimed_id);
1124
+ node_state_update.claim_id = localhost->aclk_state.claimed_id;
1125
+ query->data.bin_payload.payload = generate_node_instance_connection(&query->data.bin_payload.size, &node_state_update);
1126
rrdhost_aclk_state_unlock(localhost);
1102
- query->data.node_update.live = list->live;
1103
- query->data.node_update.hops = list->hops;
1104
- query->data.node_update.node_id = mallocz(UUID_STR_LEN);
1105
- uuid_unparse_lower(list->node_id, (char*)query->data.node_update.node_id);
1106
- query->data.node_update.queryable = 1;
1107
- query->data.node_update.session_id = aclk_session_newarch;
1108
- freez(list->hostname);
1109
- info("Queuing status update for node=%s, live=%d, hops=%d",(char*)query->data.node_update.node_id,
1127
+ info("Queuing status update for node=%s, live=%d, hops=%d",(char*)node_state_update.node_id,
1128
list->live,
1129
list->hops);
1130
+ freez((void*)node_state_update.node_id);
1131
+ query->data.bin_payload.msg_name = "UpdateNodeInstanceConnection";
1132
+ query->data.bin_payload.topic = ACLK_TOPICID_NODE_CONN;
1133
aclk_queue_query(query);
1134
} else {
1135
aclk_query_t create_query;
1136
create_query = aclk_query_new(REGISTER_NODE);
1137
+ node_instance_creation_t node_instance_creation = {
1138
+ .hops = list->hops,
1139
+ .hostname = list->hostname,
1140
+ };
1141
+ node_instance_creation.machine_guid = mallocz(UUID_STR_LEN);
1142
+ uuid_unparse_lower(list->host_id, (char*)node_instance_creation.machine_guid);
1143
+ create_query->data.bin_payload.topic = ACLK_TOPICID_CREATE_NODE;
1144
+ create_query->data.bin_payload.msg_name = "CreateNodeInstance";
1145
rrdhost_aclk_state_lock(localhost);
1117
- create_query->data.node_creation.claim_id = strdupz(localhost->aclk_state.claimed_id);
1146
+ node_instance_creation.claim_id = localhost->aclk_state.claimed_id,
1147
+ create_query->data.bin_payload.payload = generate_node_instance_creation(&create_query->data.bin_payload.size, &node_instance_creation);
1148
rrdhost_aclk_state_unlock(localhost);
1119
- create_query->data.node_creation.hops = list->hops;
1120
- create_query->data.node_creation.hostname = list->hostname;
1121
- create_query->data.node_creation.machine_guid = mallocz(UUID_STR_LEN);
1122
- uuid_unparse_lower(list->host_id, (char*)create_query->data.node_creation.machine_guid);
1123
- info("Queuing registration for host=%s, hops=%d",(char*)create_query->data.node_creation.machine_guid,
1149
+ info("Queuing registration for host=%s, hops=%d",(char*)node_instance_creation.machine_guid,
1150
list->hops);
1151
+ freez(node_instance_creation.machine_guid);
1152
aclk_queue_query(create_query);
1153
}
1154
+ freez(list->hostname);
1155
1156
list++;
1157
}
1158
freez(list_head);
1159
}
1160
+#endif
1161
1162
void aclk_send_bin_msg(char *msg, size_t msg_len, enum aclk_topics subtopic, const char *msgname)
1163
{
@@ -1208,7 +1237,7 @@ char *ng_aclk_state(void)
1237
"Protocols Supported: Legacy\n"
1238
#endif
1239
);
1211
- buffer_sprintf(wb, "Protocol Used: %s\nClaimed: ", aclk_use_new_cloud_arch ? "Protobuf" : "Legacy");
1240
+ buffer_sprintf(wb, "Protocol Used: %s\nMQTT Version: %d\nClaimed: ", aclk_use_new_cloud_arch ? "Protobuf" : "Legacy", use_mqtt_5 ? 5 : 3);
1241
1242
char *agent_id = is_agent_claimed();
1243
if (agent_id == NULL)
@@ -1408,6 +1437,9 @@ char *ng_aclk_state_json(void)
1437
tmp = json_object_new_string(aclk_use_new_cloud_arch ? "Protobuf" : "Legacy");
1438
json_object_object_add(msg, "used-cloud-protocol", tmp);
1439
1440
+ tmp = json_object_new_int(use_mqtt_5 ? 5 : 3);
1441
+ json_object_object_add(msg, "mqtt-version", tmp);
1442
+
1443
tmp = json_object_new_int(aclk_rcvd_cloud_msgs);
1444
json_object_object_add(msg, "received-app-layer-msgs", tmp);
1445
aclk/aclk.h
+2
-1
@@ -43,9 +43,10 @@ int aclk_update_chart(RRDHOST *host, char *chart_name, int create);
43
void aclk_add_collector(RRDHOST *host, const char *plugin_name, const char *module_name);
44
void aclk_del_collector(RRDHOST *host, const char *plugin_name, const char *module_name);
45
46
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
47
void aclk_host_state_update(RRDHOST *host, int cmd);
47
-
48
void aclk_send_node_instances(void);
49
+#endif
50
51
void aclk_send_bin_msg(char *msg, size_t msg_len, enum aclk_topics subtopic, const char *msgname);
52
aclk/aclk_alarm_api.c
+2
-1
@@ -24,7 +24,8 @@ void aclk_send_alarm_log_entry(struct alarm_log_entry *log_entry)
24
25
aclk_send_bin_msg(payload, payload_size, ACLK_TOPICID_ALARM_LOG, "AlarmLogEntry");
26
27
- freez(payload);
27
+ if (!use_mqtt_5)
28
+ freez(payload);
29
}
30
31
void aclk_send_provide_alarm_cfg(struct provide_alarm_configuration *cfg)
aclk/aclk_api.c
+3
@@ -16,6 +16,7 @@ int aclk_disable_runtime = 0;
16
int aclk_disable_single_updates = 0;
17
18
int aclk_stats_enabled;
19
+int use_mqtt_5 = 0;
20
21
#define ACLK_IMPL_KEY_NAME "aclk implementation"
22
@@ -68,6 +69,8 @@ struct label *add_aclk_host_labels(struct label *label) {
69
break;
70
}
71
72
+ int mqtt5 = config_get_boolean(CONFIG_SECTION_CLOUD, "mqtt5", CONFIG_BOOLEAN_NO);
73
+ label = add_label_to_list(label, "_mqtt_version", mqtt5 ? "5" : "3", LABEL_SOURCE_AUTO);
74
label = add_label_to_list(label, "_aclk_impl", "Next Generation", LABEL_SOURCE_AUTO);
75
label = add_label_to_list(label, "_aclk_proxy", proxy_str, LABEL_SOURCE_AUTO);
76
#ifdef ENABLE_NEW_CLOUD_PROTOCOL
aclk/aclk_api.h
+3
@@ -21,6 +21,7 @@ extern int aclk_stats_enabled;
21
extern int aclk_alert_reloaded;
22
23
extern int aclk_ng;
24
+extern int use_mqtt_5;
25
26
#ifdef ENABLE_ACLK
27
void *aclk_starter(void *ptr);
@@ -36,7 +37,9 @@ int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae);
37
void aclk_add_collector(RRDHOST *host, const char *plugin_name, const char *module_name);
38
void aclk_del_collector(RRDHOST *host, const char *plugin_name, const char *module_name);
39
40
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
41
void aclk_host_state_update(RRDHOST *host, int connect);
42
+#endif
43
44
#define NETDATA_ACLK_HOOK \
45
{ .name = "ACLK_Main", \
aclk/aclk_query.c
+2
-18
@@ -292,22 +292,6 @@ static int alarm_state_update_query(struct aclk_query_thread *query_thr, aclk_qu
292
}
293
294
#ifdef ENABLE_NEW_CLOUD_PROTOCOL
295
-static int register_node(struct aclk_query_thread *query_thr, aclk_query_t query) {
296
- // TODO create a pending registrations list
297
- // with some timeouts to detect registration requests that
298
- // go unanswered from the cloud
299
- aclk_generate_node_registration(query_thr->client, &query->data.node_creation);
300
- return 0;
301
-}
302
-
303
-static int node_state_update(struct aclk_query_thread *query_thr, aclk_query_t query) {
304
- // TODO create a pending registrations list
305
- // with some timeouts to detect registration requests that
306
- // go unanswered from the cloud
307
- aclk_generate_node_state_update(query_thr->client, &query->data.node_update);
308
- return 0;
309
-}
310
-
295
static int send_bin_msg(struct aclk_query_thread *query_thr, aclk_query_t query)
296
{
297
// this will be simplified when legacy support is removed
@@ -324,8 +308,8 @@ aclk_query_handler aclk_query_handlers[] = {
308
{ .type = CHART_NEW, .name = "chart_new", .fnc = chart_query },
309
{ .type = CHART_DEL, .name = "chart_delete", .fnc = info_metadata },
310
#ifdef ENABLE_NEW_CLOUD_PROTOCOL
327
- { .type = REGISTER_NODE, .name = "register_node", .fnc = register_node },
328
- { .type = NODE_STATE_UPDATE, .name = "node_state_update", .fnc = node_state_update },
311
+ { .type = REGISTER_NODE, .name = "register_node", .fnc = send_bin_msg },
312
+ { .type = NODE_STATE_UPDATE, .name = "node_state_update", .fnc = send_bin_msg },
313
{ .type = CHART_DIMS_UPDATE, .name = "chart_and_dim_update", .fnc = send_bin_msg },
314
{ .type = CHART_CONFIG_UPDATED, .name = "chart_config_updated", .fnc = send_bin_msg },
315
{ .type = CHART_RESET, .name = "reset_chart_messages", .fnc = send_bin_msg },
aclk/aclk_query_queue.c
+2
-10
@@ -121,16 +121,7 @@ void aclk_query_free(aclk_query_t query)
121
break;
122
123
case NODE_STATE_UPDATE:
124
- freez((void*)query->data.node_update.claim_id);
125
- freez((void*)query->data.node_update.node_id);
126
- break;
127
-
124
case REGISTER_NODE:
129
- freez((void*)query->data.node_creation.claim_id);
130
- freez((void*)query->data.node_creation.hostname);
131
- freez((void*)query->data.node_creation.machine_guid);
132
- break;
133
-
125
case CHART_DIMS_UPDATE:
126
case CHART_CONFIG_UPDATED:
127
case CHART_RESET:
@@ -139,7 +130,8 @@ void aclk_query_free(aclk_query_t query)
130
case ALARM_LOG_HEALTH:
131
case ALARM_PROVIDE_CFG:
132
case ALARM_SNAPSHOT:
142
- freez(query->data.bin_payload.payload);
133
+ if (!use_mqtt_5)
134
+ freez(query->data.bin_payload.payload);
135
break;
136
137
default:
aclk/aclk_query_queue.h
-2
@@ -77,8 +77,6 @@ struct aclk_query {
77
struct aclk_query_metadata metadata_alarms;
78
struct aclk_query_http_api_v2 http_api_v2;
79
struct aclk_query_chart_add_del chart_add_del;
80
- node_instance_creation_t node_creation;
81
- node_instance_connection_t node_update;
80
struct aclk_bin_payload bin_payload;
81
json_object *alarm_update;
82
} data;
aclk/aclk_rx_msgs.c
+20
-13
@@ -277,32 +277,39 @@ int create_node_instance_result(const char *msg, size_t msg_len)
277
update_node_id(&host_id, &node_id);
278
279
aclk_query_t query = aclk_query_new(NODE_STATE_UPDATE);
280
- query->data.node_update.hops = 1; //TODO - real hop count instead of hardcoded
281
- rrdhost_aclk_state_lock(localhost);
282
- query->data.node_update.claim_id = strdupz(localhost->aclk_state.claimed_id);
283
- rrdhost_aclk_state_unlock(localhost);
280
+ node_instance_connection_t node_state_update = {
281
+ .hops = 1,
282
+ .live = 0,
283
+ .queryable = 1,
284
+ .session_id = aclk_session_newarch,
285
+ .node_id = res.node_id
286
+ };
287
288
RRDHOST *host = rrdhost_find_by_guid(res.machine_guid, 0);
286
- query->data.node_update.live = 0;
287
-
289
if (host) {
290
// not all host must have RRDHOST struct created for them
291
// if they never connected during runtime of agent
292
if (host == localhost) {
292
- query->data.node_update.live = 1;
293
- query->data.node_update.hops = 0;
293
+ node_state_update.live = 1;
294
+ node_state_update.hops = 0;
295
} else {
296
netdata_mutex_lock(&host->receiver_lock);
296
- query->data.node_update.live = (host->receiver != NULL);
297
+ node_state_update.live = (host->receiver != NULL);
298
netdata_mutex_unlock(&host->receiver_lock);
298
- query->data.node_update.hops = host->system_info->hops;
299
+ node_state_update.hops = host->system_info->hops;
300
}
301
}
302
302
- query->data.node_update.node_id = res.node_id; // aclk_query_free will free it
303
- query->data.node_update.queryable = 1;
304
- query->data.node_update.session_id = aclk_session_newarch;
303
+ rrdhost_aclk_state_lock(localhost);
304
+ node_state_update.claim_id = localhost->aclk_state.claimed_id;
305
+ query->data.bin_payload.payload = generate_node_instance_connection(&query->data.bin_payload.size, &node_state_update);
306
+ rrdhost_aclk_state_unlock(localhost);
307
+
308
+ query->data.bin_payload.msg_name = "UpdateNodeInstanceConnection";
309
+ query->data.bin_payload.topic = ACLK_TOPICID_NODE_CONN;
310
+
311
aclk_queue_query(query);
312
+ freez(res.node_id);
313
freez(res.machine_guid);
314
return 0;
315
}
aclk/aclk_tx_msgs.c
+26
-40
@@ -49,7 +49,11 @@ uint16_t aclk_send_bin_message_subtopic_pid(mqtt_wss_client client, char *msg, s
49
return 0;
50
}
51
52
- mqtt_wss_publish_pid(client, topic, msg, msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
52
+ if (use_mqtt_5)
53
+ mqtt_wss_publish5(client, (char*)topic, NULL, msg, &freez, msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
54
+ else
55
+ mqtt_wss_publish_pid(client, topic, msg, msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
56
+
57
#ifdef NETDATA_INTERNAL_CHECKS
58
aclk_stats_msg_published(packet_id);
59
#endif
@@ -125,7 +129,7 @@ static int aclk_send_message_with_bin_payload(mqtt_wss_client client, json_objec
129
130
if (unlikely(!topic || topic[0] != '/')) {
131
error ("Full topic required!");
128
- return 500;
132
+ return HTTP_RESP_INTERNAL_SERVER_ERROR;
133
}
134
135
str = json_object_to_json_string_ext(msg, JSON_C_TO_STRING_PLAIN);
@@ -149,17 +153,22 @@ static int aclk_send_message_with_bin_payload(mqtt_wss_client client, json_objec
153
json_object_to_file_ext(filename, msg, JSON_C_TO_STRING_PRETTY);
154
#endif */
155
152
- rc = mqtt_wss_publish_pid_block(client, topic, payload_len ? full_msg : str, len, MQTT_WSS_PUB_QOS1, &packet_id, 5000);
153
- if (rc == MQTT_WSS_ERR_BLOCK_TIMEOUT) {
154
- error("Timeout sending binpacked message");
155
- freez(full_msg);
156
- return 503;
157
- }
158
- if (rc == MQTT_WSS_ERR_TX_BUF_TOO_SMALL) {
159
- error("Message is bigger than allowed maximum");
160
- freez(full_msg);
161
- return 403;
156
+ if (use_mqtt_5)
157
+ mqtt_wss_publish5(client, (char*)topic, NULL, (char*)(payload_len ? full_msg : str), NULL, len, MQTT_WSS_PUB_QOS1, &packet_id);
158
+ else {
159
+ rc = mqtt_wss_publish_pid_block(client, topic, payload_len ? full_msg : str, len, MQTT_WSS_PUB_QOS1, &packet_id, 5000);
160
+ if (rc == MQTT_WSS_ERR_BLOCK_TIMEOUT) {
161
+ error("Timeout sending binpacked message");
162
+ freez(full_msg);
163
+ return HTTP_RESP_BACKEND_FETCH_FAILED;
164
+ }
165
+ if (rc == MQTT_WSS_ERR_TX_BUF_TOO_SMALL) {
166
+ error("Message is bigger than allowed maximum");
167
+ freez(full_msg);
168
+ return HTTP_RESP_FORBIDDEN;
169
+ }
170
}
171
+
172
#ifdef NETDATA_INTERNAL_CHECKS
173
aclk_stats_msg_published(packet_id);
174
#endif
@@ -363,13 +372,13 @@ void aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg
372
json_object_put(msg);
373
374
switch (rc) {
366
- case 403:
375
+ case HTTP_RESP_FORBIDDEN:
376
aclk_http_msg_v2_err(client, topic, msg_id, rc, CLOUD_EC_REQ_REPLY_TOO_BIG, CLOUD_EMSG_REQ_REPLY_TOO_BIG, payload, payload_len);
377
break;
369
- case 500:
378
+ case HTTP_RESP_INTERNAL_SERVER_ERROR:
379
aclk_http_msg_v2_err(client, topic, msg_id, rc, CLOUD_EC_FAIL_TOPIC, CLOUD_EMSG_FAIL_TOPIC, payload, payload_len);
380
break;
372
- case 503:
381
+ case HTTP_RESP_BACKEND_FETCH_FAILED:
382
aclk_http_msg_v2_err(client, topic, msg_id, rc, CLOUD_EC_SND_TIMEOUT, CLOUD_EMSG_SND_TIMEOUT, payload, payload_len);
383
break;
384
}
@@ -490,7 +499,8 @@ uint16_t aclk_send_agent_connection_update(mqtt_wss_client client, int reachable
499
}
500
501
pid = aclk_send_bin_message_subtopic_pid(client, msg, len, ACLK_TOPICID_AGENT_CONN, "UpdateAgentConnection");
493
- freez(msg);
502
+ if (!use_mqtt_5)
503
+ freez(msg);
504
if (localhost->aclk_state.prev_claimed_id) {
505
freez(localhost->aclk_state.prev_claimed_id);
506
localhost->aclk_state.prev_claimed_id = NULL;
@@ -522,30 +532,6 @@ char *aclk_generate_lwt(size_t *size) {
532
533
return msg;
534
}
525
-
526
-void aclk_generate_node_registration(mqtt_wss_client client, node_instance_creation_t *node_creation) {
527
- size_t len;
528
- char *msg = generate_node_instance_creation(&len, node_creation);
529
- if (!msg) {
530
- error("Error generating nodeinstance::create::v1::CreateNodeInstance");
531
- return;
532
- }
533
-
534
- aclk_send_bin_message_subtopic_pid(client, msg, len, ACLK_TOPICID_CREATE_NODE, "CreateNodeInstance");
535
- freez(msg);
536
-}
537
-
538
-void aclk_generate_node_state_update(mqtt_wss_client client, node_instance_connection_t *node_connection) {
539
- size_t len;
540
- char *msg = generate_node_instance_connection(&len, node_connection);
541
- if (!msg) {
542
- error("Error generating nodeinstance::v1::UpdateNodeInstanceConnection");
543
- return;
544
- }
545
-
546
- aclk_send_bin_message_subtopic_pid(client, msg, len, ACLK_TOPICID_NODE_CONN, "UpdateNodeInstanceConnection");
547
- freez(msg);
548
-}
535
#endif /* ENABLE_NEW_CLOUD_PROTOCOL */
536
537
#ifndef __GNUC__
aclk/aclk_tx_msgs.h
-3
@@ -28,9 +28,6 @@ int aclk_send_app_layer_disconnect(mqtt_wss_client client, const char *message);
28
// new protobuf msgs
29
uint16_t aclk_send_agent_connection_update(mqtt_wss_client client, int reachable);
30
char *aclk_generate_lwt(size_t *size);
31
-
32
-void aclk_generate_node_registration(mqtt_wss_client client, node_instance_creation_t *node_creation);
33
-void aclk_generate_node_state_update(mqtt_wss_client client, node_instance_connection_t *node_connection);
31
#endif
32
33
#endif
aclk/schema-wrappers/node_creation.h
+3
-3
@@ -8,9 +8,9 @@ extern "C" {
8
#endif
9
10
typedef struct {
11
- const char* claim_id;
12
- const char* machine_guid;
13
- const char* hostname;
11
+ char* claim_id;
12
+ char* machine_guid;
13
+ char* hostname;
14
15
int32_t hops;
16
} node_instance_creation_t;
mqtt_websockets
+1
-1
@@ -1 +1 @@
1
-Subproject commit 2c7c2eb583abea0137e169ad3646e843d44297ef
1
+Subproject commit 288d92b36ffab3bd078334a7a87ba2b680b7852b
streaming/receiver.c
+4
-4
@@ -682,9 +682,9 @@ static int rrdpush_receive(struct receiver_state *rpt)
682
683
cd.version = rpt->stream_version;
684
685
-#if defined(ENABLE_ACLK)
685
+#if defined(ENABLE_NEW_CLOUD_PROTOCOL)
686
// in case we have cloud connection we inform cloud
687
- // new slave connected
687
+ // new child connected
688
if (netdata_cloud_setting)
689
aclk_host_state_update(rpt->host, 1);
690
#endif
@@ -696,9 +696,9 @@ static int rrdpush_receive(struct receiver_state *rpt)
696
error("STREAM %s [receive from [%s]:%s]: disconnected (completed %zu updates).", rpt->hostname, rpt->client_ip,
697
rpt->client_port, count);
698
699
-#if defined(ENABLE_ACLK)
699
+#if defined(ENABLE_NEW_CLOUD_PROTOCOL)
700
// in case we have cloud connection we inform cloud
701
- // new slave connected
701
+ // new child connected
702
if (netdata_cloud_setting)
703
aclk_host_state_update(rpt->host, 0);
704
#endif