ACLK-NG remove 'cmd' switch by message type (#11866)
* remove legacy protocl rx msg switch
Timotej S committed
Jan 11, 2022 at 14:33 UTC
e102adc6239ef10405bcda18c02adb95da902f64
3 files changed
+39
-58
aclk/aclk.c
+1
-1
@@ -222,7 +222,7 @@ static void msg_callback_old_protocol(const char *topic, const void *msg, size_t
222
return;
223
}
224
225
- aclk_handle_cloud_message(cmsg);
225
+ aclk_handle_cloud_cmd_message(cmsg);
226
}
227
228
#ifdef ENABLE_NEW_CLOUD_PROTOCOL
aclk/aclk_rx_msgs.c
+37
-56
@@ -119,7 +119,7 @@ static inline int aclk_v2_payload_get_query(const char *payload, char **query_ur
119
}\
120
ACLK_SHARED_STATE_UNLOCK;
121
122
-static int aclk_handle_cloud_request_v2(struct aclk_request *cloud_to_agent, char *raw_payload)
122
+static int aclk_handle_cloud_http_request_v2(struct aclk_request *cloud_to_agent, char *raw_payload)
123
{
124
if (!aclk_use_new_cloud_arch) {
125
HTTP_CHECK_AGENT_INITIALIZED();
@@ -172,73 +172,43 @@ error:
172
return 1;
173
}
174
175
-typedef struct aclk_incoming_msg_type{
176
- char *name;
177
- int(*fnc)(struct aclk_request *, char *);
178
-}aclk_incoming_msg_type;
179
-
180
-aclk_incoming_msg_type aclk_incoming_msg_types_compression[] = {
181
- { .name = "http", .fnc = aclk_handle_cloud_request_v2 },
182
- { .name = NULL, .fnc = NULL }
183
-};
184
-
185
-struct aclk_incoming_msg_type *aclk_incoming_msg_types = aclk_incoming_msg_types_compression;
186
-
187
-int aclk_handle_cloud_message(char *payload)
175
+int aclk_handle_cloud_cmd_message(char *payload)
176
{
177
struct aclk_request cloud_to_agent;
178
memset(&cloud_to_agent, 0, sizeof(struct aclk_request));
179
192
- if (aclk_stats_enabled) {
193
- ACLK_STATS_LOCK;
194
- aclk_metrics_per_sample.cloud_req_recvd++;
195
- ACLK_STATS_UNLOCK;
196
- }
197
-
180
if (unlikely(!payload)) {
199
- errno = 0;
200
- error("ACLK incoming message is empty");
201
- goto err_cleanup_nojson;
181
+ error_report("ACLK incoming 'cmd' message is empty");
182
+ return 1;
183
}
184
204
- debug(D_ACLK, "ACLK incoming message (%s)", payload);
185
+ debug(D_ACLK, "ACLK incoming 'cmd' message (%s)", payload);
186
187
int rc = json_parse(payload, &cloud_to_agent, cloud_to_agent_parse);
188
189
if (unlikely(rc != JSON_OK)) {
209
- errno = 0;
210
- error("Malformed json request (%s)", payload);
190
+ error_report("Malformed json request (%s)", payload);
191
goto err_cleanup;
192
}
193
194
if (!cloud_to_agent.type_id) {
215
- errno = 0;
216
- error("Cloud message is missing compulsory key \"type\"");
195
+ error_report("Cloud message is missing compulsory key \"type\"");
196
goto err_cleanup;
197
}
198
220
-
221
- for (int i = 0; aclk_incoming_msg_types[i].name; i++) {
222
- if (strcmp(cloud_to_agent.type_id, aclk_incoming_msg_types[i].name) == 0) {
223
- if (likely(!aclk_incoming_msg_types[i].fnc(&cloud_to_agent, payload))) {
224
- // in case of success handler is supposed to clean up after itself
225
- // or as in the case of aclk_handle_cloud_request take
226
- // ownership of the pointers (done to avoid copying)
227
- // see what `aclk_queue_query` parameter `internal` does
228
-
229
- // NEVER CONTINUE THIS LOOP AFTER CALLING FUNCTION!!!
230
- // msg handlers (namely aclk_handle_version_response)
231
- // can freely change what aclk_incoming_msg_types points to
232
- // so either exit or restart this for loop
233
- freez(cloud_to_agent.type_id);
234
- return 0;
235
- }
236
- goto err_cleanup;
237
- }
199
+ // Originally we were expecting to have multiple types of 'cmd' message,
200
+ // but after the new protocol was designed we will ever only have 'http'
201
+ if (strcmp(cloud_to_agent.type_id, "http")) {
202
+ error_report("Only 'http' cmd message is supported");
203
+ goto err_cleanup;
204
}
205
240
- errno = 0;
241
- error("Unknown message type from Cloud \"%s\"", cloud_to_agent.type_id);
206
+ if (likely(!aclk_handle_cloud_http_request_v2(&cloud_to_agent, payload))) {
207
+ // aclk_handle_cloud_request takes ownership of the pointers
208
+ // (to avoid copying) in case of success
209
+ freez(cloud_to_agent.type_id);
210
+ return 0;
211
+ }
212
213
err_cleanup:
214
if (cloud_to_agent.payload)
@@ -250,13 +220,6 @@ err_cleanup:
220
if (cloud_to_agent.callback_topic)
221
freez(cloud_to_agent.callback_topic);
222
253
-err_cleanup_nojson:
254
- if (aclk_stats_enabled) {
255
- ACLK_STATS_LOCK;
256
- aclk_metrics_per_sample.cloud_req_err++;
257
- ACLK_STATS_UNLOCK;
258
- }
259
-
223
return 1;
224
}
225
@@ -272,7 +235,10 @@ int handle_old_proto_cmd(const char *msg, size_t msg_len)
235
char *str = mallocz(msg_len+1);
236
memcpy(str, msg, msg_len);
237
str[msg_len] = 0;
275
- aclk_handle_cloud_message(str);
238
+ if (aclk_handle_cloud_cmd_message(str)) {
239
+ freez(str);
240
+ return 1;
241
+ }
242
freez(str);
243
return 0;
244
}
@@ -507,14 +473,29 @@ void aclk_init_rx_msg_handlers(void)
473
474
void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len)
475
{
476
+ if (aclk_stats_enabled) {
477
+ ACLK_STATS_LOCK;
478
+ aclk_metrics_per_sample.cloud_req_recvd++;
479
+ ACLK_STATS_UNLOCK;
480
+ }
481
new_cloud_rx_msg_t *msg_descriptor = find_rx_handler_by_hash(simple_hash(message_type));
482
debug(D_ACLK, "Got message named '%s' from cloud", message_type);
483
if (unlikely(!msg_descriptor)) {
484
error("Do not know how to handle message of type '%s'. Ignoring", message_type);
485
+ if (aclk_stats_enabled) {
486
+ ACLK_STATS_LOCK;
487
+ aclk_metrics_per_sample.cloud_req_err++;
488
+ ACLK_STATS_UNLOCK;
489
+ }
490
return;
491
}
492
if (msg_descriptor->fnc(msg, msg_len)) {
493
error("Error processing message of type '%s'", message_type);
494
+ if (aclk_stats_enabled) {
495
+ ACLK_STATS_LOCK;
496
+ aclk_metrics_per_sample.cloud_req_err++;
497
+ ACLK_STATS_UNLOCK;
498
+ }
499
return;
500
}
501
}
aclk/aclk_rx_msgs.h
+1
-1
@@ -8,7 +8,7 @@
8
#include "daemon/common.h"
9
#include "libnetdata/libnetdata.h"
10
11
-int aclk_handle_cloud_message(char *payload);
11
+int aclk_handle_cloud_cmd_message(char *payload);
12
13
#ifdef ENABLE_NEW_CLOUD_PROTOCOL
14
void aclk_init_rx_msg_handlers(void);