Remove option to use MQTT 3 (#13824)
* remove mqtt3 support
Timotej S committed
Oct 28, 2022 at 14:36 UTC
50be9eb132845dddd5533ef52006ffb3bffc6ead
6 files changed
+18
-64
aclk/README.md
-2
@@ -60,12 +60,10 @@ You can configure following keys in the `netdata.conf` section `[cloud]`:
60
[cloud]
61
statistics = yes
62
query thread count = 2
63
- mqtt5 = yes
63
```
64
65
- `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.
66
- `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).
68
-- `mqtt5` allows disabling the new MQTT5 implementation which is used now by default in case of issues. This option will be removed in future stable release.
67
68
## Disable the ACLK
69
aclk/aclk.c
+15
-20
@@ -32,7 +32,6 @@ int aclk_connection_counter = 0;
32
int disconnect_req = 0;
33
34
int aclk_connected = 0;
35
-int use_mqtt_5 = 0;
35
int aclk_ctx_based = 0;
36
int aclk_disable_runtime = 0;
37
int aclk_stats_enabled;
@@ -459,9 +458,9 @@ static int aclk_block_till_recon_allowed() {
458
*/
459
static int aclk_get_transport_idx(aclk_env_t *env) {
460
for (size_t i = 0; i < env->transport_count; i++) {
462
- // currently we support only MQTT 3
461
+ // currently we support only MQTT 5
462
// therefore select first transport that matches
464
- if (env->transports[i]->type == ACLK_TRP_MQTT_3_1_1) {
463
+ if (env->transports[i]->type == ACLK_TRP_MQTT_5) {
464
return i;
465
}
466
}
@@ -495,7 +494,7 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)
494
while (!netdata_exit) {
495
char *cloud_base_url = appconfig_get(&cloud_config, CONFIG_SECTION_GLOBAL, "cloud base url", NULL);
496
if (cloud_base_url == NULL) {
498
- error("Do not move the cloud base url out of post_conf_load!!");
497
+ error_report("Do not move the cloud base url out of post_conf_load!!");
498
return -1;
499
}
500
@@ -505,7 +504,7 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)
504
info("Attempting connection now");
505
memset(&base_url, 0, sizeof(url_t));
506
if (url_parse(cloud_base_url, &base_url)) {
508
- error("ACLK base URL configuration key could not be parsed. Will retry in %d seconds.", CLOUD_BASE_URL_READ_RETRY);
507
+ error_report("ACLK base URL configuration key could not be parsed. Will retry in %d seconds.", CLOUD_BASE_URL_READ_RETRY);
508
sleep(CLOUD_BASE_URL_READ_RETRY);
509
url_t_destroy(&base_url);
510
continue;
@@ -535,7 +534,7 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)
534
ret = aclk_get_env(aclk_env, base_url.host, base_url.port);
535
url_t_destroy(&base_url);
536
if (ret) {
538
- error("Failed to Get ACLK environment");
537
+ error_report("Failed to Get ACLK environment");
538
// delay handled by aclk_block_till_recon_allowed
539
continue;
540
}
@@ -549,14 +548,14 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)
548
}
549
550
if (!aclk_env_has_capa("proto")) {
552
- error ("Can't use encoding=proto without at least \"proto\" capability.");
551
+ error_report("Can't use encoding=proto without at least \"proto\" capability.");
552
continue;
553
}
554
info("New ACLK protobuf protocol negotiated successfully (/env response).");
555
556
memset(&auth_url, 0, sizeof(url_t));
557
if (url_parse(aclk_env->auth_endpoint, &auth_url)) {
559
- error("Parsing URL returned by env endpoint for authentication failed. \"%s\"", aclk_env->auth_endpoint);
558
+ error_report("Parsing URL returned by env endpoint for authentication failed. \"%s\"", aclk_env->auth_endpoint);
559
url_t_destroy(&auth_url);
560
continue;
561
}
@@ -564,7 +563,7 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)
563
ret = aclk_get_mqtt_otp(aclk_private_key, (char **)&mqtt_conn_params.clientid, (char **)&mqtt_conn_params.username, (char **)&mqtt_conn_params.password, &auth_url);
564
url_t_destroy(&auth_url);
565
if (ret) {
567
- error("Error passing Challenge/Response to get OTP");
566
+ error_report("Error passing Challenge/Response to get OTP");
567
continue;
568
}
569
@@ -573,20 +572,20 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)
572
mqtt_conn_params.will_topic = aclk_get_topic(ACLK_TOPICID_AGENT_CONN);
573
574
if (!mqtt_conn_params.will_topic) {
576
- error("Couldn't get LWT topic. Will not send LWT.");
575
+ error_report("Couldn't get LWT topic. Will not send LWT.");
576
continue;
577
}
578
579
// Do the MQTT connection
580
ret = aclk_get_transport_idx(aclk_env);
581
if (ret < 0) {
583
- error("Cloud /env endpoint didn't return any transport usable by this Agent.");
582
+ error_report("Cloud /env endpoint didn't return any transport usable by this Agent.");
583
continue;
584
}
585
586
memset(&mqtt_url, 0, sizeof(url_t));
587
if (url_parse(aclk_env->transports[ret]->endpoint, &mqtt_url)){
589
- error("Failed to parse target URL for /env trp idx %d \"%s\"", ret, aclk_env->transports[ret]->endpoint);
588
+ error_report("Failed to parse target URL for /env trp idx %d \"%s\"", ret, aclk_env->transports[ret]->endpoint);
589
url_t_destroy(&mqtt_url);
590
continue;
591
}
@@ -672,9 +671,7 @@ void *aclk_main(void *ptr)
671
if (wait_till_agent_claim_ready())
672
goto exit;
673
675
- use_mqtt_5 = config_get_boolean(CONFIG_SECTION_CLOUD, "mqtt5", CONFIG_BOOLEAN_YES);
676
-
677
- if (!(mqttwss_client = mqtt_wss_new("mqtt_wss", aclk_mqtt_wss_log_cb, msg_callback, puback_callback, use_mqtt_5))) {
674
+ if (!(mqttwss_client = mqtt_wss_new("mqtt_wss", aclk_mqtt_wss_log_cb, msg_callback, puback_callback, 1))) {
675
error("Couldn't initialize MQTT_WSS network library");
676
goto exit;
677
}
@@ -919,7 +916,7 @@ char *aclk_state(void)
916
"ACLK Version: 2\n"
917
"Protocols Supported: Protobuf\n"
918
);
922
- buffer_sprintf(wb, "Protocol Used: Protobuf\nMQTT Version: %d\nClaimed: ", use_mqtt_5 ? 5 : 3);
919
+ buffer_sprintf(wb, "Protocol Used: Protobuf\nMQTT Version: %d\nClaimed: ", 5);
920
921
char *agent_id = get_agent_claimid();
922
if (agent_id == NULL)
@@ -1072,7 +1069,7 @@ char *aclk_state_json(void)
1069
tmp = json_object_new_string("Protobuf");
1070
json_object_object_add(msg, "used-cloud-protocol", tmp);
1071
1075
- tmp = json_object_new_int(use_mqtt_5 ? 5 : 3);
1072
+ tmp = json_object_new_int(5);
1073
json_object_object_add(msg, "mqtt-version", tmp);
1074
1075
tmp = json_object_new_int(aclk_rcvd_cloud_msgs);
@@ -1171,9 +1168,7 @@ void add_aclk_host_labels(void) {
1168
break;
1169
}
1170
1174
- int mqtt5 = config_get_boolean(CONFIG_SECTION_CLOUD, "mqtt5", CONFIG_BOOLEAN_YES);
1175
-
1176
- rrdlabels_add(labels, "_mqtt_version", mqtt5 ? "5" : "3", RRDLABEL_SRC_AUTO);
1171
+ rrdlabels_add(labels, "_mqtt_version", "5", RRDLABEL_SRC_AUTO);
1172
rrdlabels_add(labels, "_aclk_proxy", proxy_str, RRDLABEL_SRC_AUTO);
1173
rrdlabels_add(labels, "_aclk_ng_new_cloud_protocol", "true", RRDLABEL_SRC_AUTO|RRDLABEL_SRC_ACLK);
1174
#else
aclk/aclk.h
-1
@@ -14,7 +14,6 @@
14
#endif /* ENABLE_ACLK */
15
16
extern int aclk_connected;
17
-extern int use_mqtt_5;
17
extern int aclk_ctx_based;
18
extern int aclk_disable_runtime;
19
extern int aclk_stats_enabled;
aclk/aclk_alarm_api.c
-3
@@ -23,9 +23,6 @@ void aclk_send_alarm_log_entry(struct alarm_log_entry *log_entry)
23
char *payload = generate_alarm_log_entry(&payload_size, log_entry);
24
25
aclk_send_bin_msg(payload, payload_size, ACLK_TOPICID_ALARM_LOG, "AlarmLogEntry");
26
-
27
- if (!use_mqtt_5)
28
- freez(payload);
26
}
27
28
void aclk_send_provide_alarm_cfg(struct provide_alarm_configuration *cfg)
aclk/aclk_query_queue.c
-16
@@ -111,22 +111,6 @@ void aclk_query_free(aclk_query_t query)
111
freez(query->data.http_api_v2.query);
112
break;
113
114
- case NODE_STATE_UPDATE:
115
- case REGISTER_NODE:
116
- case CHART_DIMS_UPDATE:
117
- case CHART_CONFIG_UPDATED:
118
- case CHART_RESET:
119
- case RETENTION_UPDATED:
120
- case UPDATE_NODE_INFO:
121
- case ALARM_LOG_HEALTH:
122
- case ALARM_PROVIDE_CFG:
123
- case ALARM_SNAPSHOT:
124
- case UPDATE_NODE_COLLECTORS:
125
- case PROTO_BIN_MESSAGE:
126
- if (!use_mqtt_5)
127
- freez(query->data.bin_payload.payload);
128
- break;
129
-
114
default:
115
break;
116
}
aclk/aclk_tx_msgs.c
+3
-22
@@ -35,10 +35,7 @@ uint16_t aclk_send_bin_message_subtopic_pid(mqtt_wss_client client, char *msg, s
35
return 0;
36
}
37
38
- if (use_mqtt_5)
39
- mqtt_wss_publish5(client, (char*)topic, NULL, msg, &freez_aclk_publish5a, msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
40
- else
41
- mqtt_wss_publish_pid(client, topic, msg, msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
38
+ mqtt_wss_publish5(client, (char*)topic, NULL, msg, &freez_aclk_publish5a, msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
39
40
#ifdef NETDATA_INTERNAL_CHECKS
41
aclk_stats_msg_published(packet_id);
@@ -64,7 +61,7 @@ static int aclk_send_message_with_bin_payload(mqtt_wss_client client, json_objec
61
uint16_t packet_id;
62
const char *str;
63
char *full_msg = NULL;
67
- int len, rc;
64
+ int len;
65
66
if (unlikely(!topic || topic[0] != '/')) {
67
error ("Full topic required!");
@@ -87,21 +84,7 @@ static int aclk_send_message_with_bin_payload(mqtt_wss_client client, json_objec
84
len += payload_len;
85
}
86
90
- if (use_mqtt_5)
91
- mqtt_wss_publish5(client, (char*)topic, NULL, (char*)(payload_len ? full_msg : str), (payload_len ? &freez_aclk_publish5b : &json_object_put_wrapper), len, MQTT_WSS_PUB_QOS1, &packet_id);
92
- else {
93
- rc = mqtt_wss_publish_pid_block(client, topic, payload_len ? full_msg : str, len, MQTT_WSS_PUB_QOS1, &packet_id, 5000);
94
- freez(full_msg);
95
- json_object_put(msg);
96
- if (rc == MQTT_WSS_ERR_BLOCK_TIMEOUT) {
97
- error("Timeout sending binpacked message");
98
- return HTTP_RESP_BACKEND_FETCH_FAILED;
99
- }
100
- if (rc == MQTT_WSS_ERR_TX_BUF_TOO_SMALL) {
101
- error("Message is bigger than allowed maximum");
102
- return HTTP_RESP_FORBIDDEN;
103
- }
104
- }
87
+ mqtt_wss_publish5(client, (char*)topic, NULL, (char*)(payload_len ? full_msg : str), (payload_len ? &freez_aclk_publish5b : &json_object_put_wrapper), len, MQTT_WSS_PUB_QOS1, &packet_id);
88
89
#ifdef NETDATA_INTERNAL_CHECKS
90
aclk_stats_msg_published(packet_id);
@@ -263,8 +246,6 @@ uint16_t aclk_send_agent_connection_update(mqtt_wss_client client, int reachable
246
}
247
248
pid = aclk_send_bin_message_subtopic_pid(client, msg, len, ACLK_TOPICID_AGENT_CONN, "UpdateAgentConnection");
266
- if (!use_mqtt_5)
267
- freez(msg);
249
if (localhost->aclk_state.prev_claimed_id) {
250
freez(localhost->aclk_state.prev_claimed_id);
251
localhost->aclk_state.prev_claimed_id = NULL;