@cryptotaxi247 / netdata-1 / commits / fa91f42a1

Handle incoming ACLK traffic asynchronously (#19436)

* Handle incoming aclk commands async * Process async send_alarm_snapshot (shouldn't be used) * Add batch processing * Function rename and change log to debug

Stelios Fragkakis committed Jan 20, 2025 at 01:47 UTC fa91f42a15470c6ed1d478cbe40f89de81d2d6b8

9 files changed +261 -105

src/aclk/aclk.c

+4 -4

@@ -953,7 +953,7 @@ void aclk_host_state_update(RRDHOST *host, int cmd, int queryable)

953 "Registering host=%s, hops=%d", host->machine_guid,

954 rrdhost_ingestion_hops(host));

955

956 - aclk_execute_query(create_query);

956 + aclk_add_job(create_query);

957 return;

958 }

959 }

@@ -982,7 +982,7 @@ void aclk_host_state_update(RRDHOST *host, int cmd, int queryable)

982 freez((void*)node_state_update.node_id);

983 query->data.bin_payload.msg_name = "UpdateNodeInstanceConnection";

984 query->data.bin_payload.topic = ACLK_TOPICID_NODE_CONN;

985 - aclk_execute_query(query);

985 + aclk_add_job(query);

986 }

987

988 void aclk_send_node_instances()

@@ -1028,7 +1028,7 @@ void aclk_send_node_instances()

1028 freez((void*)node_state_update.node_id);

1029 query->data.bin_payload.msg_name = "UpdateNodeInstanceConnection";

1030 query->data.bin_payload.topic = ACLK_TOPICID_NODE_CONN;

1031 - aclk_execute_query(query);

1031 + aclk_add_job(query);

1032 } else {

1033 aclk_query_t create_query;

1034 create_query = aclk_query_new(REGISTER_NODE);

@@ -1050,7 +1050,7 @@ void aclk_send_node_instances()

1050 (char*)node_instance_creation.machine_guid, list->hops);

1051

1052 freez((void *)node_instance_creation.machine_guid);

1053 - aclk_execute_query(create_query);

1053 + aclk_add_job(create_query);

1054 }

1055 freez(list->hostname);

1056

src/aclk/aclk_query_queue.h

+10 -2

@@ -23,6 +23,9 @@ typedef enum {

23 CTX_CHECKPOINT, // Context checkpoint from the cloud

24 CTX_STOP_STREAMING, // Context stop streaming

25 CREATE_NODE_INSTANCE, // Create node instance on the agent

26 + SEND_NODE_INSTANCES, // Send node instances to the cloud

27 + ALERT_START_STREAMING, // Start alert streaming from cloud

28 + ALERT_CHECKPOINT, // Do an alert version check

29 ACLK_QUERY_TYPE_COUNT // always keep this as last

30 } aclk_query_type_t;

@@ -50,17 +53,21 @@ struct aclk_query {

53 char *dedup_id;

54 char *callback_topic;

55 char *msg_id;

56 + union {

57 + char *claim_id;

58 + char *machine_guid;

59 + };

61 struct timeval created_tv;

62 usec_t created;

63 int timeout;

58 - // TODO maybe remove?

59 - int version;

65 + uint64_t version;

66 union {

67 struct aclk_query_http_api_v2 http_api_v2;

68 struct aclk_bin_payload bin_payload;

69 void *payload;

70 + char *node_id;

71 } data;

72 };

@@ -68,6 +75,7 @@ aclk_query_t aclk_query_new(aclk_query_type_t type);

75 void aclk_query_free(aclk_query_t query);

77 void aclk_execute_query(aclk_query_t query);

78 +void aclk_add_job(aclk_query_t query);

80 #define QUEUE_IF_PAYLOAD_PRESENT(query) \

81 do { \

src/aclk/aclk_rx_msgs.c

+33 -59

@@ -255,49 +255,11 @@ int create_node_instance_result(const char *msg, size_t msg_len)

255

256 netdata_log_debug(D_ACLK, "CreateNodeInstanceResult: guid:%s nodeid:%s", res.machine_guid, res.node_id);

257

258 - nd_uuid_t host_id, node_id;

259 - if (uuid_parse(res.machine_guid, host_id)) {

260 - netdata_log_error("Error parsing machine_guid provided by CreateNodeInstanceResult");

261 - freez(res.machine_guid);

262 - freez(res.node_id);

263 - return 1;

264 - }

265 - if (uuid_parse(res.node_id, node_id)) {

266 - netdata_log_error("Error parsing node_id provided by CreateNodeInstanceResult");

267 - freez(res.machine_guid);

268 - freez(res.node_id);

269 - return 1;

270 - }

271 - sql_update_node_id(&host_id, &node_id);

272 -

273 - aclk_query_t query = aclk_query_new(NODE_STATE_UPDATE);

274 - node_instance_connection_t node_state_update = {

275 - .hops = 1,

276 - .live = 0,

277 - .queryable = 1,

278 - .session_id = aclk_session_newarch,

279 - .node_id = res.node_id,

280 - .capabilities = NULL

281 - };

282 -

283 - RRDHOST *host = rrdhost_find_by_guid(res.machine_guid);

284 - if (likely(host)) {

285 - node_state_update.live = rrdhost_is_local(host) ? 1 : 0;

286 - node_state_update.hops = rrdhost_ingestion_hops(host);

287 - node_state_update.capabilities = aclk_get_node_instance_capas(host);

288 - schedule_node_state_update(host, 5000);

289 - }

290 -

291 - CLAIM_ID claim_id = claim_id_get();

292 - node_state_update.claim_id = claim_id_is_set(claim_id) ? claim_id.str : NULL;

293 - query->data.bin_payload.payload = generate_node_instance_connection(&query->data.bin_payload.size, &node_state_update);

294 -

295 - freez((void *)node_state_update.capabilities);

296 -

297 - query->data.bin_payload.msg_name = "UpdateNodeInstanceConnection";

298 - query->data.bin_payload.topic = ACLK_TOPICID_NODE_CONN;

258 + aclk_query_t query = aclk_query_new(CREATE_NODE_INSTANCE);

259 + query->data.node_id = strdupz(res.node_id);

260 + query->machine_guid = strdupz(res.machine_guid);

261 + aclk_add_job(query);

262

300 - aclk_execute_query(query);

263 freez(res.node_id);

264 freez(res.machine_guid);

265 return 0;

@@ -307,7 +269,8 @@ int send_node_instances(const char *msg, size_t msg_len)

269 {

270 UNUSED(msg);

271 UNUSED(msg_len);

310 - aclk_send_node_instances();

272 + aclk_query_t query = aclk_query_new(SEND_NODE_INSTANCES);

273 + aclk_add_job(query);

274 return 0;

275 }

276

@@ -342,7 +305,10 @@ int start_alarm_streaming(const char *msg, size_t msg_len)

305 netdata_log_error("Error parsing StartAlarmStreaming");

306 return 1;

307 }

345 - aclk_start_alert_streaming(res.node_id, res.version);

308 + aclk_query_t query = aclk_query_new(ALERT_START_STREAMING);

309 + query->data.node_id = strdupz(res.node_id);

310 + query->version = res.version;

311 + aclk_add_job(query);

312 freez(res.node_id);

313 return 0;

314 }

@@ -356,7 +322,11 @@ int send_alarm_checkpoint(const char *msg, size_t msg_len)

322 freez(sac.claim_id);

323 return 1;

324 }

359 - aclk_alert_version_check(sac.node_id, sac.claim_id, sac.version);

325 + aclk_query_t query = aclk_query_new(ALERT_CHECKPOINT);

326 + query->data.node_id = strdupz(sac.node_id);

327 + query->claim_id = strdupz(sac.claim_id);

328 + query->version = sac.version;

329 + aclk_add_job(query);

330 freez(sac.node_id);

331 freez(sac.claim_id);

332 return 0;

@@ -383,7 +353,11 @@ int send_alarm_snapshot(const char *msg, size_t msg_len)

353 destroy_send_alarm_snapshot(sas);

354 return 1;

355 }

386 - aclk_process_send_alarm_snapshot(sas->node_id, sas->claim_id, sas->snapshot_uuid);

356 + aclk_query_t query = aclk_query_new(ALERT_CHECKPOINT);

357 + query->data.node_id = strdupz(sas->node_id);

358 + query->claim_id = strdupz(sas->claim_id);

359 + query->version = 0; // force snapshot

360 + aclk_add_job(query);

361 destroy_send_alarm_snapshot(sas);

362 return 0;

363 }

@@ -420,7 +394,7 @@ int contexts_checkpoint(const char *msg, size_t msg_len)

394

395 aclk_query_t query = aclk_query_new(CTX_CHECKPOINT);

396 query->data.payload = cmd;

423 - aclk_execute_query(query);

397 + aclk_add_job(query);

398 return 0;

399 }

400

@@ -437,7 +411,7 @@ int stop_streaming_contexts(const char *msg, size_t msg_len)

411

412 aclk_query_t query = aclk_query_new(CTX_STOP_STREAMING);

413 query->data.payload = cmd;

440 - aclk_execute_query(query);

414 + aclk_add_job(query);

415 return 0;

416 }

417

@@ -466,18 +440,18 @@ typedef struct {

440

441 new_cloud_rx_msg_t rx_msgs[] = {

442 { .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 = "SendAlarmCheckpoint", .name_hash = 0, .fnc = send_alarm_checkpoint },

476 - { .name = "SendAlarmConfiguration", .name_hash = 0, .fnc = send_alarm_configuration },

477 - { .name = "SendAlarmSnapshot", .name_hash = 0, .fnc = send_alarm_snapshot },

443 + { .name = "CreateNodeInstanceResult", .name_hash = 0, .fnc = create_node_instance_result }, // async

444 + { .name = "SendNodeInstances", .name_hash = 0, .fnc = send_node_instances }, // async

445 + { .name = "StreamChartsAndDimensions", .name_hash = 0, .fnc = stream_charts_and_dimensions }, // unused

446 + { .name = "ChartsAndDimensionsAck", .name_hash = 0, .fnc = charts_and_dimensions_ack }, // unused

447 + { .name = "UpdateChartConfigs", .name_hash = 0, .fnc = update_chart_configs }, // unused

448 + { .name = "StartAlarmStreaming", .name_hash = 0, .fnc = start_alarm_streaming }, // async

449 + { .name = "SendAlarmCheckpoint", .name_hash = 0, .fnc = send_alarm_checkpoint }, // async

450 + { .name = "SendAlarmConfiguration", .name_hash = 0, .fnc = send_alarm_configuration }, // async

451 + { .name = "SendAlarmSnapshot", .name_hash = 0, .fnc = send_alarm_snapshot }, // shouldn't be used

452 { .name = "DisconnectReq", .name_hash = 0, .fnc = handle_disconnect_req },

479 - { .name = "ContextsCheckpoint", .name_hash = 0, .fnc = contexts_checkpoint },

480 - { .name = "StopStreamingContexts", .name_hash = 0, .fnc = stop_streaming_contexts },

453 + { .name = "ContextsCheckpoint", .name_hash = 0, .fnc = contexts_checkpoint }, // async

454 + { .name = "StopStreamingContexts", .name_hash = 0, .fnc = stop_streaming_contexts }, // async

455 { .name = "CancelPendingRequest", .name_hash = 0, .fnc = cancel_pending_req },

456 { .name = NULL, .name_hash = 0, .fnc = NULL },

457 };

src/daemon/libuv_workers.c

@@ -71,6 +71,10 @@ void register_libuv_worker_jobs() {

71 worker_register_job_name(UV_EVENT_CTX_SEND_SNAPSHOT, "ctx send snapshot");

72 worker_register_job_name(UV_EVENT_CTX_SEND_SNAPSHOT_UPD, "ctx send update");

73 worker_register_job_name(UV_EVENT_NODE_STATE_UPDATE, "node state update");

74 + worker_register_job_name(UV_EVENT_SEND_NODE_INSTANCES, "send node instances");

75 + worker_register_job_name(UV_EVENT_ALERT_START_STREAMING, "alert start streaming");

76 + worker_register_job_name(UV_EVENT_ALERT_CHECKPOINT, "alert checkpoint");

77 + worker_register_job_name(UV_EVENT_CREATE_NODE_INSTANCE, "create node instance");

79 // netdatacli

80 worker_register_job_name(UV_EVENT_SCHEDULE_CMD, "schedule command");

src/daemon/libuv_workers.h

@@ -64,6 +64,10 @@ enum event_loop_job {

64 UV_EVENT_CTX_SEND_SNAPSHOT,

65 UV_EVENT_CTX_SEND_SNAPSHOT_UPD,

66 UV_EVENT_NODE_STATE_UPDATE,

67 + UV_EVENT_SEND_NODE_INSTANCES,

68 + UV_EVENT_ALERT_START_STREAMING,

69 + UV_EVENT_ALERT_CHECKPOINT,

70 + UV_EVENT_CREATE_NODE_INSTANCE,

72 // netdatacli

73 UV_EVENT_SCHEDULE_CMD,

src/database/sqlite/sqlite_aclk.c

+204 -9

@@ -11,6 +11,55 @@ void sanity_check(void) {

11 #include "sqlite_aclk_node.h"

12 #include "../aclk_query_queue.h"

13 #include "../aclk_query.h"

14 +#include "../aclk_capas.h"

15 +

16 +static void create_node_instance_result_job(const char *machine_guid, const char *node_id)

17 +{

18 + nd_uuid_t host_uuid, node_uuid;

19 +

20 + if (uuid_parse(machine_guid, host_uuid)) {

21 + netdata_log_error("Error parsing machine_guid provided by CreateNodeInstanceResult");

22 + return;

23 + }

24 +

25 + if (uuid_parse(node_id, node_uuid)) {

26 + netdata_log_error("Error parsing node_id provided by CreateNodeInstanceResult");

27 + return;

28 + }

29 +

30 + RRDHOST *host = rrdhost_find_by_guid(machine_guid);

31 + if (unlikely(!host)) {

32 + netdata_log_error("Cannot find machine_guid provided by CreateNodeInstanceResult");

33 + return;

34 + }

35 +

36 + sql_update_node_id(&host_uuid, &node_uuid);

37 +

38 + aclk_query_t query = aclk_query_new(NODE_STATE_UPDATE);

39 + node_instance_connection_t node_state_update = {

40 + .hops = 1,

41 + .live = 0,

42 + .queryable = 1,

43 + .session_id = aclk_session_newarch,

44 + .node_id = node_id,

45 + .capabilities = NULL};

46 +

47 + node_state_update.live = rrdhost_is_local(host) ? 1 : 0;

48 + node_state_update.hops = rrdhost_ingestion_hops(host);

49 + node_state_update.capabilities = aclk_get_node_instance_capas(host);

50 + schedule_node_state_update(host, 5000);

51 +

52 + CLAIM_ID claim_id = claim_id_get();

53 + node_state_update.claim_id = claim_id_is_set(claim_id) ? claim_id.str : NULL;

54 + query->data.bin_payload.payload = generate_node_instance_connection(&query->data.bin_payload.size, &node_state_update);

55 +

56 + freez((void *)node_state_update.capabilities);

57 +

58 + query->data.bin_payload.msg_name = "UpdateNodeInstanceConnection";

59 + query->data.bin_payload.topic = ACLK_TOPICID_NODE_CONN;

60 +

61 + aclk_add_job(query);

62 +}

64 struct aclk_sync_config_s {

65 uv_thread_t thread;

@@ -21,7 +70,9 @@ struct aclk_sync_config_s {

70 mqtt_wss_client client;

71 int aclk_queries_running;

72 bool alert_push_running;

73 + bool aclk_batch_job_is_running;

74 SPINLOCK cmd_queue_lock;

75 + uint32_t aclk_jobs_pending;

76 struct aclk_database_cmd *cmd_base;

77 } aclk_sync_config = { 0 };

@@ -269,6 +320,11 @@ static int aclk_config_parameters(void *data __maybe_unused, int argc __maybe_un

320 return 0;

321 }

322

323 +struct judy_list_t {

324 + Pvoid_t JudyL;

325 + Word_t count;

326 +};

327 +

328 static void async_cb(uv_async_t *handle)

329 {

330 uv_stop(handle->loop);

@@ -281,12 +337,18 @@ static void timer_cb(uv_timer_t *handle)

337 {

338 uv_stop(handle->loop);

339 uv_update_time(handle->loop);

340 + struct aclk_sync_config_s *config = handle->data;

341

342 struct aclk_database_cmd cmd = { 0 };

343 if (aclk_online_for_alerts()) {

344 cmd.opcode = ACLK_DATABASE_PUSH_ALERT;

345 aclk_database_enq_cmd(&cmd);

346 }

347 +

348 + if (config->aclk_jobs_pending > 0) {

349 + cmd.opcode = ACLK_QUERY_BATCH_EXECUTE;

350 + aclk_database_enq_cmd(&cmd);

351 + }

352 }

353

354 struct aclk_query_payload {

@@ -315,6 +377,8 @@ static void aclk_run_query(struct aclk_sync_config_s *config, aclk_query_t query

377 bool ok_to_send = true;

378

379 switch (query->type) {

380 +

381 +// Incoming : cloud -> agent

382 case HTTP_API_V2:

383 if (is_worker)

384 worker_is_busy(UV_EVENT_ACLK_QUERY_EXECUTE);

@@ -341,6 +405,37 @@ static void aclk_run_query(struct aclk_sync_config_s *config, aclk_query_t query

405 freez(cmd);

406 ok_to_send = false;

407 break;

408 + case SEND_NODE_INSTANCES:

409 + if (is_worker)

410 + worker_is_busy(UV_EVENT_SEND_NODE_INSTANCES);

411 + aclk_send_node_instances();

412 + ok_to_send = false;

413 + break;

414 + case ALERT_START_STREAMING:

415 + if (is_worker)

416 + worker_is_busy(UV_EVENT_ALERT_START_STREAMING);

417 + aclk_start_alert_streaming(query->data.node_id, query->version);

418 + freez(query->data.node_id);

419 + ok_to_send = false;

420 + break;

421 + case ALERT_CHECKPOINT:

422 + if (is_worker)

423 + worker_is_busy(UV_EVENT_ALERT_CHECKPOINT);

424 + aclk_alert_version_check(query->data.node_id, query->claim_id, query->version);

425 + freez(query->data.node_id);

426 + freez(query->claim_id);

427 + ok_to_send = false;

428 + break;

429 + case CREATE_NODE_INSTANCE:

430 + if (is_worker)

431 + worker_is_busy(UV_EVENT_CREATE_NODE_INSTANCE);

432 + create_node_instance_result_job(query->machine_guid, query->data.node_id);

433 + freez(query->data.node_id);

434 + freez(query->machine_guid);

435 + ok_to_send = false;

436 + break;

437 +

438 +// Outgoing: agent -> cloud

439 case ALARM_PROVIDE_CFG:

440 if (is_worker)

441 worker_is_busy(UV_EVENT_ALARM_PROVIDE_CFG);

@@ -378,8 +473,10 @@ static void aclk_run_query(struct aclk_sync_config_s *config, aclk_query_t query

473 ok_to_send = false;

474 break;

475 }

476 +

477 if (ok_to_send)

478 send_bin_msg(config->client, query);

479 +

480 aclk_query_free(query);

481 }

482

@@ -395,6 +492,51 @@ static void aclk_run_query_job(uv_work_t *req)

492 worker_is_idle();

493 }

494

495 +static void after_aclk_execute_batch(uv_work_t *req, int status __maybe_unused)

496 +{

497 + struct aclk_query_payload *payload = req->data;

498 + struct aclk_sync_config_s *config = payload->config;

499 + config->aclk_batch_job_is_running = false;

500 + freez(payload);

501 +}

502 +

503 +static void aclk_execute_batch(uv_work_t *req)

504 +{

505 + register_libuv_worker_jobs();

506 +

507 + struct aclk_query_payload *payload = req->data;

508 + struct aclk_sync_config_s *config = payload->config;

509 + struct judy_list_t *aclk_query_batch = payload->data;

510 +

511 + if (!aclk_query_batch)

512 + return;

513 +

514 + usec_t started_ut = now_monotonic_usec(); (void)started_ut;

515 +

516 + size_t entries = aclk_query_batch->count;

517 + Word_t Index = 0;

518 + bool first = true;

519 + Pvoid_t *PValue;

520 + while ((PValue = JudyLFirstThenNext(aclk_query_batch->JudyL, &Index, &first))) {

521 + if (!*PValue)

522 + continue;

523 +

524 + aclk_query_t query = *PValue;

525 + aclk_run_query(config, query, true);

526 + }

527 +

528 + (void) JudyLFreeArray(&aclk_query_batch->JudyL, PJE0);

529 + freez(aclk_query_batch);

530 +

531 + usec_t ended_ut = now_monotonic_usec();

532 + (void)ended_ut;

533 + nd_log_daemon(

534 + NDLP_DEBUG, "Processed %zu ACLK commands in %0.2f ms", entries, (double)(ended_ut - started_ut) / USEC_PER_MS);

535 +

536 + worker_is_idle();

537 +}

538 +

539 +

540 static void node_update_timer_cb(uv_timer_t *handle)

541 {

542 struct aclk_sync_cfg_t *ahc = handle->data;

@@ -441,6 +583,8 @@ static void start_alert_push(uv_work_t *req __maybe_unused)

583 worker_is_idle();

584 }

585

586 +#define MAX_ACLK_BATCH_JOBS_IN_QUEUE (20)

587 +

588 static void aclk_synchronization(void *arg)

589 {

590 struct aclk_sync_config_s *config = arg;

@@ -448,11 +592,13 @@ static void aclk_synchronization(void *arg)

592 worker_register("ACLKSYNC");

593 service_register(SERVICE_THREAD_TYPE_EVENT_LOOP, NULL, NULL, NULL, true);

594

451 - worker_register_job_name(ACLK_DATABASE_NOOP, "noop");

452 - worker_register_job_name(ACLK_DATABASE_NODE_STATE, "node state");

453 - worker_register_job_name(ACLK_DATABASE_PUSH_ALERT, "alert push");

454 - worker_register_job_name(ACLK_DATABASE_PUSH_ALERT_CONFIG, "alert conf push");

455 - worker_register_job_name(ACLK_QUERY_EXECUTE_SYNC, "aclk query execute sync");

595 + worker_register_job_name(ACLK_DATABASE_NOOP, "noop");

596 + worker_register_job_name(ACLK_DATABASE_NODE_STATE, "node state");

597 + worker_register_job_name(ACLK_DATABASE_PUSH_ALERT, "alert push");

598 + worker_register_job_name(ACLK_DATABASE_PUSH_ALERT_CONFIG, "alert conf push");

599 + worker_register_job_name(ACLK_QUERY_EXECUTE_SYNC, "aclk query execute sync");

600 + worker_register_job_name(ACLK_QUERY_BATCH_EXECUTE, "aclk batch execute");

601 + worker_register_job_name(ACLK_QUERY_BATCH_ADD, "aclk batch add");

602

603 uv_loop_t *loop = &config->loop;

604 fatal_assert(0 == uv_loop_init(loop));

@@ -472,6 +618,10 @@ static void aclk_synchronization(void *arg)

618 netdata_log_info("Starting ACLK synchronization thread with %d parallel query threads", query_thread_count);

619

620 struct alert_push_data *data;

621 + aclk_query_t query;

622 + struct judy_list_t *aclk_query_batch = NULL;

623 + Pvoid_t *Pvalue;

624 + struct aclk_query_payload *payload;

625

626 while (likely(service_running(SERVICE_ACLK))) {

627 enum aclk_database_opcode opcode;

@@ -555,10 +705,9 @@ static void aclk_synchronization(void *arg)

705 config->client = (mqtt_wss_client) cmd.param[0];

706 break;

707

558 - case ACLK_QUERY_EXECUTE:;

559 - aclk_query_t query = (aclk_query_t)cmd.param[0];

560 -

561 - struct aclk_query_payload *payload = NULL;

708 + case ACLK_QUERY_EXECUTE:

709 + query = (aclk_query_t)cmd.param[0];

710 + payload = NULL;

711 config->aclk_queries_running++;

712 bool execute_now = (config->aclk_queries_running > query_thread_count);

713 if (!execute_now) {

@@ -577,6 +726,44 @@ static void aclk_synchronization(void *arg)

726 }

727 break;

728

729 +// Note: The following two opcodes must be in this order

730 + case ACLK_QUERY_BATCH_ADD:

731 + query = (aclk_query_t)cmd.param[0];

732 + if (!query)

733 + break;

734 +

735 + if (!aclk_query_batch)

736 + aclk_query_batch = callocz(1, sizeof(*aclk_query_batch));

737 +

738 + Pvalue = JudyLIns(&aclk_query_batch->JudyL, ++aclk_query_batch->count, PJE0);

739 + if (Pvalue)

740 + *Pvalue = query;

741 +

742 + config->aclk_jobs_pending++;

743 + if (aclk_query_batch->count < MAX_ACLK_BATCH_JOBS_IN_QUEUE || config->aclk_batch_job_is_running)

744 + break;

745 + // fall through

746 + case ACLK_QUERY_BATCH_EXECUTE:

747 + if (!aclk_query_batch || config->aclk_batch_job_is_running)

748 + break;

749 +

750 + payload = mallocz(sizeof(*payload));

751 + payload->request.data = payload;

752 + payload->config = config;

753 + payload->data = aclk_query_batch;

754 +

755 + config->aclk_batch_job_is_running = true;

756 + config->aclk_jobs_pending -= aclk_query_batch->count;

757 + aclk_query_batch = NULL;

758 +

759 + if (uv_queue_work(loop, &payload->request, aclk_execute_batch, after_aclk_execute_batch)) {

760 + aclk_query_batch = payload->data;

761 + config->aclk_jobs_pending += aclk_query_batch->count;

762 + freez(payload);

763 + config->aclk_batch_job_is_running = false;

764 + }

765 + break;

766 +

767 default:

768 break;

769 }

@@ -702,6 +889,14 @@ void aclk_execute_query(aclk_query_t query)

889 queue_aclk_sync_cmd(ACLK_QUERY_EXECUTE, query, NULL);

890 }

891

892 +void aclk_add_job(aclk_query_t query)

893 +{

894 + if (unlikely(!aclk_sync_config.initialized))

895 + return;

896 +

897 + queue_aclk_sync_cmd(ACLK_QUERY_BATCH_ADD, query, NULL);

898 +}

899 +

900 void aclk_query_init(mqtt_wss_client client) {

901

902 queue_aclk_sync_cmd(ACLK_MQTT_WSS_CLIENT, client, NULL);

src/database/sqlite/sqlite_aclk.h

@@ -24,6 +24,8 @@ enum aclk_database_opcode {

24 ACLK_MQTT_WSS_CLIENT,

25 ACLK_QUERY_EXECUTE,

26 ACLK_QUERY_EXECUTE_SYNC,

27 + ACLK_QUERY_BATCH_ADD,

28 + ACLK_QUERY_BATCH_EXECUTE,

30 // leave this last

31 // we need it to check for worker utilization

src/database/sqlite/sqlite_aclk_alert.c

-30

@@ -819,36 +819,6 @@ static void schedule_alert_snapshot_if_needed(struct aclk_sync_cfg_t *wc, uint64

819 wc->checkpoint_count++;

820 }

821

822 -void aclk_process_send_alarm_snapshot(char *node_id, char *claim_id __maybe_unused, char *snapshot_uuid)

823 -{

824 - nd_uuid_t node_uuid;

825 -

826 - if (unlikely(!node_id || uuid_parse(node_id, node_uuid)))

827 - return;

828 -

829 - struct aclk_sync_cfg_t *wc;

830 -

831 - RRDHOST *host = find_host_by_node_id(node_id);

832 - if (unlikely(!host || !(wc = host->aclk_config))) {

833 - nd_log(NDLS_ACCESS, NDLP_WARNING, "ACLK STA [%s (N/A)]: ACLK node id does not exist", node_id);

834 - return;

835 - }

836 -

837 - nd_log(NDLS_ACCESS, NDLP_DEBUG,

838 - "IN [%s (%s)]: Request to send alerts snapshot, snapshot_uuid %s",

839 - node_id,

840 - wc->host ? rrdhost_hostname(wc->host) : "N/A",

841 - snapshot_uuid);

842 -

843 - if (wc->alerts_snapshot_uuid && !strcmp(wc->alerts_snapshot_uuid,snapshot_uuid))

844 - return;

845 -

846 - wc->alerts_snapshot_uuid = strdupz(snapshot_uuid);

847 -

848 - wc->send_snapshot = 1;

849 - rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);

850 -}

851 -

822 #define SQL_COUNT_SNAPSHOT_ENTRIES \

823 "SELECT COUNT(1) FROM alert_version av, health_log hl " \

824 "WHERE hl.host_id = @host_id AND hl.health_log_id = av.health_log_id AND av.status <> -2"

src/database/sqlite/sqlite_aclk_alert.h

-1

@@ -11,7 +11,6 @@ void aclk_start_alert_streaming(char *node_id, uint64_t cloud_version);

11 void aclk_alert_version_check(char *node_id, char *claim_id, uint64_t cloud_version);

13 void send_alert_snapshot_to_cloud(RRDHOST *host __maybe_unused);

14 -void aclk_process_send_alarm_snapshot(char *node_id, char *claim_id, char *snapshot_uuid);

14 bool process_alert_pending_queue(RRDHOST *host);

15 void aclk_push_alert_events_for_all_hosts(void);

16 uint64_t calculate_node_alert_version(RRDHOST *host);