@cryptotaxi247 / netdata-1 / commits / 2298e8b82

Further improve alert processing (#19489)

More pre compiled statements (compile once and use in a loop before finalize) Create a new index on alert_queue

Stelios Fragkakis committed Jan 27, 2025 at 10:21 UTC 2298e8b8259973f7616bb054666045b24d5c233c

3 files changed +57 -30

src/database/sqlite/sqlite_aclk_alert.c

+40 -27

@@ -262,7 +262,7 @@ static inline char *sqlite3_text_strdupz_empty(sqlite3_stmt *res, int iCol) {

262

263 #define SQL_UPDATE_ALERT_VERSION \

264 "INSERT INTO alert_version (health_log_id, unique_id, status, version, date_submitted)" \

265 - " VALUES (@health_log_id, @unique_id, @status, @version, UNIXEPOCH())" \

265 + " VALUES (@health_log_id, @unique_id, @status, @version, UNIXEPOCH())" \

266 " ON CONFLICT(health_log_id) DO UPDATE SET status = excluded.status, version = excluded.version, " \

267 " unique_id=excluded.unique_id, date_submitted=excluded.date_submitted"

268

@@ -270,27 +270,32 @@ static inline char *sqlite3_text_strdupz_empty(sqlite3_stmt *res, int iCol) {

270 // Store a new alert transition along with the version after sending to the cloud

271 // - Update an existing alert with the updated version, status, transition and date submitted

272 //

273 -static void sql_update_alert_version(int64_t health_log_id, int64_t unique_id, RRDCALC_STATUS status, uint64_t version)

273 +static void sql_update_alert_version(

274 + int64_t health_log_id,

275 + int64_t unique_id,

276 + RRDCALC_STATUS status,

277 + uint64_t version,

278 + sqlite3_stmt **res)

279 {

275 - sqlite3_stmt *res = NULL;

276 -

277 - if (!PREPARE_STATEMENT(db_meta, SQL_UPDATE_ALERT_VERSION, &res))

278 - return;

280 + if (!*res) {

281 + if (!PREPARE_STATEMENT(db_meta, SQL_UPDATE_ALERT_VERSION, res))

282 + return;

283 + }

284

285 int param = 0;

281 - SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, health_log_id));

282 - SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, unique_id));

283 - SQLITE_BIND_FAIL(done, sqlite3_bind_int(res, ++param, status));

284 - SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, version));

286 + SQLITE_BIND_FAIL(done, sqlite3_bind_int64(*res, ++param, health_log_id));

287 + SQLITE_BIND_FAIL(done, sqlite3_bind_int64(*res, ++param, unique_id));

288 + SQLITE_BIND_FAIL(done, sqlite3_bind_int(*res, ++param, status));

289 + SQLITE_BIND_FAIL(done, sqlite3_bind_int64(*res, ++param, version));

290

291 param = 0;

287 - int rc = sqlite3_step_monitored(res);

292 + int rc = sqlite3_step_monitored(*res);

293 if (rc != SQLITE_DONE)

294 error_report("Failed to execute sql_update_alert_version");

295

296 done:

292 - REPORT_BIND_FAIL(res, param);

293 - SQLITE_FINALIZE(res);

297 + REPORT_BIND_FAIL(*res, param);

298 + SQLITE_RESET(*res);

299 }

300

301 #define SQL_SELECT_ALERT_TO_DUMMY \

@@ -318,6 +323,7 @@ static void commit_alert_events(RRDHOST *host)

323 int64_t first_sequence_id = 0;

324 int64_t last_sequence_id = 0;

325

326 + sqlite3_stmt *res_version = NULL;

327 param = 0;

328 while (sqlite3_step_monitored(res) == SQLITE_ROW) {

329

@@ -330,7 +336,9 @@ static void commit_alert_events(RRDHOST *host)

336 RRDCALC_STATUS status = (RRDCALC_STATUS)sqlite3_column_int(res, 3);

337 int64_t health_log_id = sqlite3_column_int64(res, 4);

338

333 - sql_update_alert_version(health_log_id, unique_id, status, version);

339 + // Prepare the statement on the first time (res_version) then reuse it

340 + // finalize when we are done

341 + sql_update_alert_version(health_log_id, unique_id, status, version, &res_version);

342 }

343

344 if (first_sequence_id)

@@ -469,21 +477,20 @@ void health_alarm_log_populate(

477 " AND hl.host_id = @host_id AND aq.host_id = hl.host_id AND hl.health_log_id = hld.health_log_id" \

478 " ORDER BY aq.sequence_id ASC LIMIT "ACLK_MAX_ALERT_UPDATES

479

472 -static void aclk_push_alert_event(RRDHOST *host __maybe_unused)

473 -

480 +static void aclk_push_alert_event(RRDHOST *host, sqlite3_stmt **res, sqlite3_stmt **res_version)

481 {

482 CLAIM_ID claim_id = claim_id_get();

483

484 if (!claim_id_is_set(claim_id) || UUIDiszero(host->node_id))

485 return;

486

480 - sqlite3_stmt *res = NULL;

481 -

482 - if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_ALERT_TO_PUSH, &res))

483 - return;

487 + if (!*res) {

488 + if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_ALERT_TO_PUSH, res))

489 + return;

490 + }

491

492 int param = 0;

486 - SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));

493 + SQLITE_BIND_FAIL(done, sqlite3_bind_blob(*res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));

494

495 char node_id_str[UUID_STR_LEN];

496 uuid_unparse_lower(host->node_id.uuid, node_id_str);

@@ -498,8 +505,8 @@ static void aclk_push_alert_event(RRDHOST *host __maybe_unused)

505 param = 0;

506 RRDCALC_STATUS status;

507 struct aclk_sync_cfg_t *wc = host->aclk_config;

501 - while (sqlite3_step_monitored(res) == SQLITE_ROW) {

502 - health_alarm_log_populate(&alarm_log, res, host, &status);

508 + while (sqlite3_step_monitored(*res) == SQLITE_ROW) {

509 + health_alarm_log_populate(&alarm_log, *res, host, &status);

510 aclk_send_alarm_log_entry(&alarm_log);

511 wc->alert_count++;

512

@@ -507,7 +514,9 @@ static void aclk_push_alert_event(RRDHOST *host __maybe_unused)

514 if (first_id == 0)

515 first_id = last_id;

516

510 - sql_update_alert_version(alarm_log.health_log_id, alarm_log.unique_id, status, alarm_log.version);

517 + // The statement to set the version will be compiled once and reset when done

518 + // out caller will finalize the statement to release resources

519 + sql_update_alert_version(alarm_log.health_log_id, alarm_log.unique_id, status, alarm_log.version, res_version);

520

521 destroy_alarm_log_entry(&alarm_log);

522 }

@@ -528,8 +537,8 @@ static void aclk_push_alert_event(RRDHOST *host __maybe_unused)

537 }

538

539 done:

531 - REPORT_BIND_FAIL(res, param);

532 - SQLITE_FINALIZE(res);

540 + REPORT_BIND_FAIL(*res, param);

541 + SQLITE_RESET(*res);

542 }

543

544 #define SQL_DELETE_PROCESSED_ROWS "DELETE FROM alert_queue WHERE host_id = @host_id AND rowid = @row"

@@ -668,6 +677,8 @@ void aclk_push_alert_events_for_all_hosts(void)

677 {

678 RRDHOST *host;

679

680 + sqlite3_stmt *res = NULL; // used to scan pending alerts to send

681 + sqlite3_stmt *res_version = NULL; // used to update the alert version

682 dfe_start_reentrant(rrdhost_root_index, host) {

683 if (!rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS) ||

684 rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))

@@ -694,9 +705,11 @@ void aclk_push_alert_events_for_all_hosts(void)

705 wc->send_snapshot = 0;

706 }

707 else

697 - aclk_push_alert_event(host);

708 + aclk_push_alert_event(host, &res, &res_version);

709 }

710 dfe_done(host);

711 + SQLITE_FINALIZE(res);

712 + SQLITE_FINALIZE(res_version);

713 }

714

715 void aclk_send_alert_configuration(char *config_hash)

src/database/sqlite/sqlite_health.c

+15 -3

@@ -939,10 +939,19 @@ done:

939 int sql_health_get_last_executed_event(RRDHOST *host, ALARM_ENTRY *ae, RRDCALC_STATUS *last_executed_status)

940 {

941 int ret = -1;

942 + static __thread sqlite3_stmt *compiled_res = NULL;

943 sqlite3_stmt *res = NULL;

944

944 - if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_HEALTH_LAST_EXECUTED_EVENT, &res))

945 - return ret;

945 + if (is_health_thread) {

946 + if (!compiled_res) {

947 + if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_SELECT_HEALTH_LAST_EXECUTED_EVENT, &compiled_res))

948 + return ret;

949 + }

950 + res = compiled_res;

951 + } else {

952 + if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_HEALTH_LAST_EXECUTED_EVENT, &res))

953 + return ret;

954 + }

955

956 int param = 0;

957 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));

@@ -959,7 +968,10 @@ int sql_health_get_last_executed_event(RRDHOST *host, ALARM_ENTRY *ae, RRDCALC_S

968

969 done:

970 REPORT_BIND_FAIL(res, param);

962 - SQLITE_FINALIZE(res);

971 + if (is_health_thread)

972 + SQLITE_RESET(res);

973 + else

974 + SQLITE_FINALIZE(res);

975 return ret;

976 }

977

src/database/sqlite/sqlite_metadata.c

@@ -89,6 +89,8 @@ const char *database_config[] = {

89 " (host_id BLOB, health_log_id INT, unique_id INT, alarm_id INT, status INT, date_scheduled INT, "

90 " UNIQUE(host_id, health_log_id, alarm_id))",

92 + "CREATE INDEX IF NOT EXISTS ind_alert_queue1 ON alert_queue(host_id, date_scheduled)",

93 +

94 "CREATE TABLE IF NOT EXISTS alert_version (health_log_id INTEGER PRIMARY KEY, unique_id INT, status INT, "

95 "version INT, date_submitted INT)",