Optimize rx msg name resolution (#11811)
* optimize rx msg name resolution
Timotej S committed
Jan 4, 2022 at 11:01 UTC
fa0045202fe4438fc090479913dc34338c3b5d11
3 files changed
+237
-151
aclk/aclk.c
+4
@@ -763,6 +763,10 @@ void *aclk_main(void *ptr)
763
return NULL;
764
}
765
766
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
767
+ aclk_init_rx_msg_handlers();
768
+#endif
769
+
770
// This thread is unusual in that it cannot be cancelled by cancel_main_threads()
771
// as it must notify the far end that it shutdown gracefully and avoid the LWT.
772
netdata_thread_disable_cancelability();
aclk/aclk_rx_msgs.c
+232
-151
@@ -261,180 +261,261 @@ err_cleanup_nojson:
261
}
262
263
#ifdef ENABLE_NEW_CLOUD_PROTOCOL
264
-void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len)
264
+typedef uint32_t simple_hash_t;
265
+typedef int(*rx_msg_handler)(const char *msg, size_t msg_len);
266
+
267
+int handle_old_proto_cmd(const char *msg, size_t msg_len)
268
{
266
- // TODO do the look up table with hashes to optimize when there are more
267
- // than few
268
- if (!strcmp(message_type, "cmd")) {
269
- // msg is binary payload in all other cases
270
- // however in this message from old legacy cloud
271
- // we have to convert it to C string
272
- char *str = mallocz(msg_len+1);
273
- memcpy(str, msg, msg_len);
274
- str[msg_len] = 0;
275
- aclk_handle_cloud_message(str);
276
- freez(str);
277
- return;
278
- }
279
- if (!strcmp(message_type, "CreateNodeInstanceResult")) {
280
- node_instance_creation_result_t res = parse_create_node_instance_result(msg, msg_len);
281
- if (!res.machine_guid || !res.node_id) {
282
- error_report("Error parsing CreateNodeInstanceResult");
283
- freez(res.machine_guid);
284
- freez(res.node_id);
285
- return;
286
- }
269
+ // msg is binary payload in all other cases
270
+ // however in this message from old legacy cloud
271
+ // we have to convert it to C string
272
+ char *str = mallocz(msg_len+1);
273
+ memcpy(str, msg, msg_len);
274
+ str[msg_len] = 0;
275
+ aclk_handle_cloud_message(str);
276
+ freez(str);
277
+ return 0;
278
+}
279
288
- debug(D_ACLK, "CreateNodeInstanceResult: guid:%s nodeid:%s", res.machine_guid, res.node_id);
280
+int create_node_instance_result(const char *msg, size_t msg_len)
281
+{
282
+ node_instance_creation_result_t res = parse_create_node_instance_result(msg, msg_len);
283
+ if (!res.machine_guid || !res.node_id) {
284
+ error_report("Error parsing CreateNodeInstanceResult");
285
+ freez(res.machine_guid);
286
+ freez(res.node_id);
287
+ return 1;
288
+ }
289
290
- uuid_t host_id, node_id;
291
- if (uuid_parse(res.machine_guid, host_id)) {
292
- error("Error parsing machine_guid provided by CreateNodeInstanceResult");
293
- freez(res.machine_guid);
294
- freez(res.node_id);
295
- return;
296
- }
297
- if (uuid_parse(res.node_id, node_id)) {
298
- error("Error parsing node_id provided by CreateNodeInstanceResult");
299
- freez(res.machine_guid);
300
- freez(res.node_id);
301
- return;
302
- }
303
- update_node_id(&host_id, &node_id);
304
-
305
- aclk_query_t query = aclk_query_new(NODE_STATE_UPDATE);
306
- query->data.node_update.hops = 1; //TODO - real hop count instead of hardcoded
307
- rrdhost_aclk_state_lock(localhost);
308
- query->data.node_update.claim_id = strdupz(localhost->aclk_state.claimed_id);
309
- rrdhost_aclk_state_unlock(localhost);
310
-
311
- RRDHOST *host = rrdhost_find_by_guid(res.machine_guid, 0);
312
- query->data.node_update.live = 0;
313
-
314
- if (host) {
315
- // not all host must have RRDHOST struct created for them
316
- // if they never connected during runtime of agent
317
- if (host == localhost) {
318
- query->data.node_update.live = 1;
319
- query->data.node_update.hops = 0;
320
- } else {
321
- netdata_mutex_lock(&host->receiver_lock);
322
- query->data.node_update.live = (host->receiver != NULL);
323
- netdata_mutex_unlock(&host->receiver_lock);
324
- query->data.node_update.hops = host->system_info->hops;
325
- }
326
- }
290
+ debug(D_ACLK, "CreateNodeInstanceResult: guid:%s nodeid:%s", res.machine_guid, res.node_id);
291
328
- query->data.node_update.node_id = res.node_id; // aclk_query_free will free it
329
- query->data.node_update.queryable = 1;
330
- query->data.node_update.session_id = aclk_session_newarch;
331
- aclk_queue_query(query);
292
+ uuid_t host_id, node_id;
293
+ if (uuid_parse(res.machine_guid, host_id)) {
294
+ error("Error parsing machine_guid provided by CreateNodeInstanceResult");
295
freez(res.machine_guid);
333
- return;
296
+ freez(res.node_id);
297
+ return 1;
298
}
335
- if (!strcmp(message_type, "SendNodeInstances")) {
336
- debug(D_ACLK, "Got SendNodeInstances");
337
- aclk_send_node_instances();
338
- return;
299
+ if (uuid_parse(res.node_id, node_id)) {
300
+ error("Error parsing node_id provided by CreateNodeInstanceResult");
301
+ freez(res.machine_guid);
302
+ freez(res.node_id);
303
+ return 1;
304
}
340
-
341
- if (!strcmp(message_type, "StreamChartsAndDimensions")) {
342
- stream_charts_and_dims_t res = parse_stream_charts_and_dims(msg, msg_len);
343
- if (!res.claim_id || !res.node_id) {
344
- error("Error parsing StreamChartsAndDimensions msg");
345
- freez(res.claim_id);
346
- freez(res.node_id);
347
- return;
305
+ update_node_id(&host_id, &node_id);
306
+
307
+ aclk_query_t query = aclk_query_new(NODE_STATE_UPDATE);
308
+ query->data.node_update.hops = 1; //TODO - real hop count instead of hardcoded
309
+ rrdhost_aclk_state_lock(localhost);
310
+ query->data.node_update.claim_id = strdupz(localhost->aclk_state.claimed_id);
311
+ rrdhost_aclk_state_unlock(localhost);
312
+
313
+ RRDHOST *host = rrdhost_find_by_guid(res.machine_guid, 0);
314
+ query->data.node_update.live = 0;
315
+
316
+ if (host) {
317
+ // not all host must have RRDHOST struct created for them
318
+ // if they never connected during runtime of agent
319
+ if (host == localhost) {
320
+ query->data.node_update.live = 1;
321
+ query->data.node_update.hops = 0;
322
+ } else {
323
+ netdata_mutex_lock(&host->receiver_lock);
324
+ query->data.node_update.live = (host->receiver != NULL);
325
+ netdata_mutex_unlock(&host->receiver_lock);
326
+ query->data.node_update.hops = host->system_info->hops;
327
}
349
- chart_batch_id = res.batch_id;
350
- aclk_start_streaming(res.node_id, res.seq_id, res.seq_id_created_at.tv_sec, res.batch_id);
328
+ }
329
+
330
+ query->data.node_update.node_id = res.node_id; // aclk_query_free will free it
331
+ query->data.node_update.queryable = 1;
332
+ query->data.node_update.session_id = aclk_session_newarch;
333
+ aclk_queue_query(query);
334
+ freez(res.machine_guid);
335
+ return 0;
336
+}
337
+
338
+int send_node_instances(const char *msg, size_t msg_len)
339
+{
340
+ UNUSED(msg);
341
+ UNUSED(msg_len);
342
+ aclk_send_node_instances();
343
+ return 0;
344
+}
345
+
346
+int stream_charts_and_dimensions(const char *msg, size_t msg_len)
347
+{
348
+ stream_charts_and_dims_t res = parse_stream_charts_and_dims(msg, msg_len);
349
+ if (!res.claim_id || !res.node_id) {
350
+ error("Error parsing StreamChartsAndDimensions msg");
351
freez(res.claim_id);
352
freez(res.node_id);
353
- return;
353
+ return 1;
354
}
355
- if (!strcmp(message_type, "ChartsAndDimensionsAck")) {
356
- chart_and_dim_ack_t res = parse_chart_and_dimensions_ack(msg, msg_len);
357
- if (!res.claim_id || !res.node_id) {
358
- error("Error parsing StreamChartsAndDimensions msg");
359
- freez(res.claim_id);
360
- freez(res.node_id);
361
- return;
362
- }
363
- aclk_ack_chart_sequence_id(res.node_id, res.last_seq_id);
355
+ chart_batch_id = res.batch_id;
356
+ aclk_start_streaming(res.node_id, res.seq_id, res.seq_id_created_at.tv_sec, res.batch_id);
357
+ freez(res.claim_id);
358
+ freez(res.node_id);
359
+ return 0;
360
+}
361
+
362
+int charts_and_dimensions_ack(const char *msg, size_t msg_len)
363
+{
364
+ chart_and_dim_ack_t res = parse_chart_and_dimensions_ack(msg, msg_len);
365
+ if (!res.claim_id || !res.node_id) {
366
+ error("Error parsing StreamChartsAndDimensions msg");
367
freez(res.claim_id);
368
freez(res.node_id);
366
- return;
367
- }
368
- if (!strcmp(message_type, "UpdateChartConfigs")) {
369
- struct update_chart_config res = parse_update_chart_config(msg, msg_len);
370
- if (!res.claim_id || !res.node_id || !res.hashes)
371
- error("Error parsing UpdateChartConfigs msg");
372
- else
373
- aclk_get_chart_config(res.hashes);
374
- destroy_update_chart_config(&res);
375
- return;
369
+ return 1;
370
}
377
- if (!strcmp(message_type, "StartAlarmStreaming")) {
378
- struct start_alarm_streaming res = parse_start_alarm_streaming(msg, msg_len);
379
- if (!res.node_id || !res.batch_id) {
380
- error("Error parsing StartAlarmStreaming");
381
- freez(res.node_id);
382
- return;
383
- }
384
- aclk_start_alert_streaming(res.node_id, res.batch_id, res.start_seq_id);
371
+ aclk_ack_chart_sequence_id(res.node_id, res.last_seq_id);
372
+ freez(res.claim_id);
373
+ freez(res.node_id);
374
+ return 0;
375
+}
376
+
377
+int update_chart_configs(const char *msg, size_t msg_len)
378
+{
379
+ struct update_chart_config res = parse_update_chart_config(msg, msg_len);
380
+ if (!res.claim_id || !res.node_id || !res.hashes)
381
+ error("Error parsing UpdateChartConfigs msg");
382
+ else
383
+ aclk_get_chart_config(res.hashes);
384
+ destroy_update_chart_config(&res);
385
+ return 0;
386
+}
387
+
388
+int start_alarm_streaming(const char *msg, size_t msg_len)
389
+{
390
+ struct start_alarm_streaming res = parse_start_alarm_streaming(msg, msg_len);
391
+ if (!res.node_id || !res.batch_id) {
392
+ error("Error parsing StartAlarmStreaming");
393
freez(res.node_id);
386
- return;
394
+ return 1;
395
}
388
- if (!strcmp(message_type, "SendAlarmLogHealth")) {
389
- char *node_id = parse_send_alarm_log_health(msg, msg_len);
390
- if (!node_id) {
391
- error("Error parsing SendAlarmLogHealth");
392
- return;
393
- }
394
- aclk_send_alarm_health_log(node_id);
395
- freez(node_id);
396
- return;
396
+ aclk_start_alert_streaming(res.node_id, res.batch_id, res.start_seq_id);
397
+ freez(res.node_id);
398
+ return 0;
399
+}
400
+
401
+int send_alarm_log_health(const char *msg, size_t msg_len)
402
+{
403
+ char *node_id = parse_send_alarm_log_health(msg, msg_len);
404
+ if (!node_id) {
405
+ error("Error parsing SendAlarmLogHealth");
406
+ return 1;
407
}
398
- if (!strcmp(message_type, "SendAlarmConfiguration")) {
399
- char *config_hash = parse_send_alarm_configuration(msg, msg_len);
400
- if (!config_hash || !*config_hash) {
401
- error("Error parsing SendAlarmConfiguration");
402
- freez(config_hash);
403
- return;
404
- }
405
- aclk_send_alarm_configuration(config_hash);
408
+ aclk_send_alarm_health_log(node_id);
409
+ freez(node_id);
410
+ return 0;
411
+}
412
+
413
+int send_alarm_configuration(const char *msg, size_t msg_len)
414
+{
415
+ char *config_hash = parse_send_alarm_configuration(msg, msg_len);
416
+ if (!config_hash || !*config_hash) {
417
+ error("Error parsing SendAlarmConfiguration");
418
freez(config_hash);
407
- return;
419
+ return 1;
420
}
409
- if (!strcmp(message_type, "SendAlarmSnapshot")) {
410
- struct send_alarm_snapshot *sas = parse_send_alarm_snapshot(msg, msg_len);
411
- if (!sas->node_id || !sas->claim_id) {
412
- error("Error parsing SendAlarmSnapshot");
413
- destroy_send_alarm_snapshot(sas);
414
- return;
415
- }
416
- aclk_process_send_alarm_snapshot(sas->node_id, sas->claim_id, sas->snapshot_id, sas->sequence_id);
421
+ aclk_send_alarm_configuration(config_hash);
422
+ freez(config_hash);
423
+ return 0;
424
+}
425
+
426
+int send_alarm_snapshot(const char *msg, size_t msg_len)
427
+{
428
+ struct send_alarm_snapshot *sas = parse_send_alarm_snapshot(msg, msg_len);
429
+ if (!sas->node_id || !sas->claim_id) {
430
+ error("Error parsing SendAlarmSnapshot");
431
destroy_send_alarm_snapshot(sas);
418
- return;
432
+ return 1;
433
}
420
- if (!strcmp(message_type, "DisconnectReq")) {
421
- struct disconnect_cmd *cmd = parse_disconnect_cmd(msg, msg_len);
422
- if (!cmd)
423
- return;
424
- if (cmd->permaban) {
425
- error ("Cloud Banned This Agent!");
426
- aclk_disable_runtime = 1;
427
- }
428
- info ("Cloud requested disconnect (EC=%u, \"%s\")", (unsigned int)cmd->error_code, cmd->error_description);
429
- if (cmd->reconnect_after_s > 0) {
430
- aclk_block_until = now_monotonic_sec() + cmd->reconnect_after_s;
431
- info ("Cloud asks not to reconnect for %u seconds. We shall honor that request", (unsigned int)cmd->reconnect_after_s);
434
+ aclk_process_send_alarm_snapshot(sas->node_id, sas->claim_id, sas->snapshot_id, sas->sequence_id);
435
+ destroy_send_alarm_snapshot(sas);
436
+ return 0;
437
+}
438
+
439
+int handle_disconnect_req(const char *msg, size_t msg_len)
440
+{
441
+ struct disconnect_cmd *cmd = parse_disconnect_cmd(msg, msg_len);
442
+ if (!cmd)
443
+ return 1;
444
+ if (cmd->permaban) {
445
+ error("Cloud Banned This Agent!");
446
+ aclk_disable_runtime = 1;
447
+ }
448
+ info("Cloud requested disconnect (EC=%u, \"%s\")", (unsigned int)cmd->error_code, cmd->error_description);
449
+ if (cmd->reconnect_after_s > 0) {
450
+ aclk_block_until = now_monotonic_sec() + cmd->reconnect_after_s;
451
+ info(
452
+ "Cloud asks not to reconnect for %u seconds. We shall honor that request",
453
+ (unsigned int)cmd->reconnect_after_s);
454
+ }
455
+ disconnect_req = 1;
456
+ freez(cmd->error_description);
457
+ freez(cmd);
458
+ return 0;
459
+}
460
+
461
+typedef struct {
462
+ const char *name;
463
+ simple_hash_t name_hash;
464
+ rx_msg_handler fnc;
465
+} new_cloud_rx_msg_t;
466
+
467
+new_cloud_rx_msg_t rx_msgs[] = {
468
+ { .name = "cmd", .name_hash = 0, .fnc = handle_old_proto_cmd },
469
+ { .name = "CreateNodeInstanceResult", .name_hash = 0, .fnc = create_node_instance_result },
470
+ { .name = "SendNodeInstances", .name_hash = 0, .fnc = send_node_instances },
471
+ { .name = "StreamChartsAndDimensions", .name_hash = 0, .fnc = stream_charts_and_dimensions },
472
+ { .name = "ChartsAndDimensionsAck", .name_hash = 0, .fnc = charts_and_dimensions_ack },
473
+ { .name = "UpdateChartConfigs", .name_hash = 0, .fnc = update_chart_configs },
474
+ { .name = "StartAlarmStreaming", .name_hash = 0, .fnc = start_alarm_streaming },
475
+ { .name = "SendAlarmLogHealth", .name_hash = 0, .fnc = send_alarm_log_health },
476
+ { .name = "SendAlarmConfiguration", .name_hash = 0, .fnc = send_alarm_configuration },
477
+ { .name = "SendAlarmSnapshot", .name_hash = 0, .fnc = send_alarm_snapshot },
478
+ { .name = "DisconnectReq", .name_hash = 0, .fnc = handle_disconnect_req },
479
+ { .name = NULL, .name_hash = 0, .fnc = NULL },
480
+};
481
+
482
+new_cloud_rx_msg_t *find_rx_handler_by_hash(simple_hash_t hash)
483
+{
484
+ // we can afford to not compare strings after hash match
485
+ // because we check for collisions at initialization in
486
+ // aclk_init_rx_msg_handlers()
487
+ for (int i = 0; rx_msgs[i].fnc; i++) {
488
+ if (rx_msgs[i].name_hash == hash)
489
+ return &rx_msgs[i];
490
+ }
491
+ return NULL;
492
+}
493
+
494
+void aclk_init_rx_msg_handlers(void)
495
+{
496
+ for (int i = 0; rx_msgs[i].fnc; i++) {
497
+ simple_hash_t hash = simple_hash(rx_msgs[i].name);
498
+ new_cloud_rx_msg_t *hdl = find_rx_handler_by_hash(hash);
499
+ if (unlikely(hdl)) {
500
+ // the list of message names changes only by changing
501
+ // the source code, therefore fatal is appropriate
502
+ fatal("Hash collision. Choose better hash. Added '%s' clashes with existing '%s'", rx_msgs[i].name, hdl->name);
503
}
433
- disconnect_req = 1;
434
- freez(cmd->error_description);
435
- freez(cmd);
504
+ rx_msgs[i].name_hash = hash;
505
+ }
506
+}
507
+
508
+void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len)
509
+{
510
+ new_cloud_rx_msg_t *msg_descriptor = find_rx_handler_by_hash(simple_hash(message_type));
511
+ debug(D_ACLK, "Got message named '%s' from cloud", message_type);
512
+ if (unlikely(!msg_descriptor)) {
513
+ error("Do not know how to handle message of type '%s'. Ignoring", message_type);
514
+ return;
515
+ }
516
+ if (msg_descriptor->fnc(msg, msg_len)) {
517
+ error("Error processing message of type '%s'", message_type);
518
return;
519
}
438
- error ("Unknown new cloud arch message type received \"%s\"", message_type);
520
}
521
#endif
aclk/aclk_rx_msgs.h
+1
@@ -11,6 +11,7 @@
11
int aclk_handle_cloud_message(char *payload);
12
13
#ifdef ENABLE_NEW_CLOUD_PROTOCOL
14
+void aclk_init_rx_msg_handlers(void);
15
void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len);
16
#endif
17