@cryptotaxi247 / netdata-1 / commits / bb775105f

Improve alert transition processing (#19487)

Use prepared statements when processing alerts from the health thread

Stelios Fragkakis committed Jan 26, 2025 at 18:53 UTC bb775105ffee2f8a62e7f685e5860963f27c6fd5

3 files changed +170 -40

src/database/sqlite/sqlite_aclk_alert.c

+93 -19

@@ -5,6 +5,8 @@

6 #include "../../aclk/aclk_alarm_api.h"

8 +extern __thread bool is_health_thread;

9 +

10 #define SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param) \

11 ({ \

12 int _param = (param); \

@@ -18,10 +20,19 @@

21 static inline bool is_event_from_alert_variable_config(int64_t unique_id, nd_uuid_t *host_id)

22 {

23 + static __thread sqlite3_stmt *compiled_res = NULL;

24 sqlite3_stmt *res = NULL;

23 - if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_VARIABLE_ALERT_BY_UNIQUE_ID, &res))

24 - return false;

26 + if (is_health_thread) {

27 + if (!compiled_res) {

28 + if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_SELECT_VARIABLE_ALERT_BY_UNIQUE_ID, &compiled_res))

29 + return false;

30 + }

31 + res = compiled_res;

32 + } else {

33 + if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_VARIABLE_ALERT_BY_UNIQUE_ID, &res))

34 + return false;

35 + }

37 bool ret = false;

@@ -34,7 +45,10 @@ static inline bool is_event_from_alert_variable_config(int64_t unique_id, nd_uui

46 done:

47 REPORT_BIND_FAIL(res, param);

37 - SQLITE_FINALIZE(res);

48 + if (is_health_thread)

49 + SQLITE_RESET(res);

50 + else

51 + SQLITE_FINALIZE(res);

52 return ret;

53 }

@@ -43,10 +57,19 @@ done:

58 static void update_alert_version_transition(int64_t health_log_id, int64_t unique_id)

59 {

60 + static __thread sqlite3_stmt *compiled_res = NULL;

61 sqlite3_stmt *res = NULL;

48 - if (!PREPARE_STATEMENT(db_meta, SQL_UPDATE_ALERT_VERSION_TRANSITION, &res))

49 - return;

63 + if (is_health_thread) {

64 + if (!compiled_res) {

65 + if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_UPDATE_ALERT_VERSION_TRANSITION, &compiled_res))

66 + return;

67 + }

68 + res = compiled_res;

69 + } else {

70 + if (!PREPARE_STATEMENT(db_meta, SQL_UPDATE_ALERT_VERSION_TRANSITION, &res))

71 + return;

72 + }

74 int param = 0;

75 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, unique_id));

@@ -59,7 +82,10 @@ static void update_alert_version_transition(int64_t health_log_id, int64_t uniqu

83 done:

84 REPORT_BIND_FAIL(res, param);

62 - SQLITE_FINALIZE(res);

85 + if (is_health_thread)

86 + SQLITE_RESET(res);

87 + else

88 + SQLITE_FINALIZE(res);

89 }

91 //decide if some events should be sent or not

@@ -68,10 +94,19 @@ done:

95 static bool cloud_status_matches(int64_t health_log_id, RRDCALC_STATUS status)

96 {

97 + static __thread sqlite3_stmt *compiled_res = NULL;

98 sqlite3_stmt *res = NULL;

73 - if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_LAST_ALERT_STATUS, &res))

74 - return true;

100 + if (is_health_thread) {

101 + if (!compiled_res) {

102 + if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_SELECT_LAST_ALERT_STATUS, &compiled_res))

103 + return true;

104 + }

105 + res = compiled_res;

106 + } else {

107 + if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_LAST_ALERT_STATUS, &res))

108 + return true;

109 + }

110

111 bool send = false;

112

@@ -87,7 +122,10 @@ static bool cloud_status_matches(int64_t health_log_id, RRDCALC_STATUS status)

122

123 done:

124 REPORT_BIND_FAIL(res, param);

90 - SQLITE_FINALIZE(res);

125 + if (is_health_thread)

126 + SQLITE_RESET(res);

127 + else

128 + SQLITE_FINALIZE(res);

129 return send;

130 }

131

@@ -106,6 +144,7 @@ done:

144 //

145 static int insert_alert_to_submit_queue(RRDHOST *host, int64_t health_log_id, uint32_t unique_id, RRDCALC_STATUS status)

146 {

147 + static __thread sqlite3_stmt *compiled_res = NULL;

148 sqlite3_stmt *res = NULL;

149

150 if (cloud_status_matches(health_log_id, status)) {

@@ -116,8 +155,16 @@ static int insert_alert_to_submit_queue(RRDHOST *host, int64_t health_log_id, ui

155 if (is_event_from_alert_variable_config(unique_id, &host->host_id.uuid))

156 return 2;

157

119 - if (!PREPARE_STATEMENT(db_meta, SQL_QUEUE_ALERT_TO_CLOUD, &res))

120 - return -1;

158 + if (is_health_thread) {

159 + if (!compiled_res) {

160 + if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_QUEUE_ALERT_TO_CLOUD, &compiled_res))

161 + return -1;

162 + }

163 + res = compiled_res;

164 + } else {

165 + if (!PREPARE_STATEMENT(db_meta, SQL_QUEUE_ALERT_TO_CLOUD, &res))

166 + return -1;

167 + }

168

169 int param = 0;

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

@@ -125,13 +172,16 @@ static int insert_alert_to_submit_queue(RRDHOST *host, int64_t health_log_id, ui

172 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, (int64_t) unique_id));

173

174 param = 0;

128 - int rc = execute_insert(res);

175 + int rc = sqlite3_step_monitored(res);

176 if (unlikely(rc != SQLITE_DONE))

177 error_report("Failed to insert alert in the submit queue %"PRIu32", rc = %d", unique_id, rc);

178

179 done:

180 REPORT_BIND_FAIL(res, param);

134 - SQLITE_FINALIZE(res);

181 + if (is_health_thread)

182 + SQLITE_RESET(res);

183 + else

184 + SQLITE_FINALIZE(res);

185 return 0;

186 }

187

@@ -486,10 +536,19 @@ done:

536

537 static void delete_alert_from_pending_queue(RRDHOST *host, int64_t row)

538 {

539 + static __thread sqlite3_stmt *compiled_res = NULL;

540 sqlite3_stmt *res = NULL;

541

491 - if (!PREPARE_STATEMENT(db_meta, SQL_DELETE_PROCESSED_ROWS, &res))

492 - return;

542 + if (is_health_thread) {

543 + if (!compiled_res) {

544 + if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_DELETE_PROCESSED_ROWS, &compiled_res))

545 + return;

546 + }

547 + res = compiled_res;

548 + } else {

549 + if (!PREPARE_STATEMENT(db_meta, SQL_DELETE_PROCESSED_ROWS, &res))

550 + return;

551 + }

552

553 int param = 0;

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

@@ -502,7 +561,10 @@ static void delete_alert_from_pending_queue(RRDHOST *host, int64_t row)

561

562 done:

563 REPORT_BIND_FAIL(res, param);

505 - SQLITE_FINALIZE(res);

564 + if (is_health_thread)

565 + SQLITE_RESET(res);

566 + else

567 + SQLITE_FINALIZE(res);

568 }

569

570 #define SQL_REBUILD_HOST_ALERT_VERSION_TABLE \

@@ -554,10 +616,19 @@ done:

616

617 bool process_alert_pending_queue(RRDHOST *host)

618 {

619 + static __thread sqlite3_stmt *compiled_res = NULL;

620 sqlite3_stmt *res = NULL;

621

559 - if (!PREPARE_STATEMENT(db_meta, SQL_PROCESS_ALERT_PENDING_QUEUE, &res))

560 - return false;

622 + if (is_health_thread) {

623 + if (!compiled_res) {

624 + if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_PROCESS_ALERT_PENDING_QUEUE, &compiled_res))

625 + return false;

626 + }

627 + res = compiled_res;

628 + } else {

629 + if (!PREPARE_STATEMENT(db_meta, SQL_PROCESS_ALERT_PENDING_QUEUE, &res))

630 + return false;

631 + }

632

633 int param = 0;

634 int added =0, count = 0;

@@ -586,7 +657,10 @@ bool process_alert_pending_queue(RRDHOST *host)

657 nd_log(NDLS_ACCESS, NDLP_NOTICE, "ACLK STA [%s (N/A)]: Processed %d entries, queued %d", rrdhost_hostname(host), count, added);

658 done:

659 REPORT_BIND_FAIL(res, param);

589 - SQLITE_FINALIZE(res);

660 + if (is_health_thread)

661 + SQLITE_RESET(res);

662 + else

663 + SQLITE_FINALIZE(res);

664 return added > 0;

665 }

666

src/database/sqlite/sqlite_health.c

+75 -21

@@ -6,6 +6,8 @@

6 #include "health/health_internals.h"

7 #include "health/health-alert-entry.h"

9 +extern __thread bool is_health_thread;

10 +

11 #define MAX_HEALTH_SQL_SIZE 2048

12 #define SQLITE3_BIND_STRING_OR_NULL(res, param, key) \

13 ((key) ? sqlite3_bind_text((res), (param), string2str(key), -1, SQLITE_STATIC) : sqlite3_bind_null((res), (param)))

@@ -27,12 +29,21 @@

30 static void sql_health_alarm_log_update(RRDHOST *host, ALARM_ENTRY *ae)

31 {

30 -

32 + static __thread sqlite3_stmt *compiled_res = NULL;

33 sqlite3_stmt *res = NULL;

32 - int rc;

34 - if (!PREPARE_STATEMENT(db_meta, SQL_UPDATE_HEALTH_LOG, &res))

35 - return;

35 + if (is_health_thread) {

36 + if (!compiled_res) {

37 + if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_UPDATE_HEALTH_LOG, &compiled_res))

38 + return;

39 + }

40 + res = compiled_res;

41 + } else {

42 + if (!PREPARE_STATEMENT(db_meta, SQL_UPDATE_HEALTH_LOG, &res))

43 + return;

44 + }

45 +

46 + int rc;

48 int param = 0;

49 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, (sqlite3_int64) ae->updated_by_id));

@@ -44,14 +55,17 @@ static void sql_health_alarm_log_update(RRDHOST *host, ALARM_ENTRY *ae)

55 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &ae->transition_id, sizeof(ae->transition_id), SQLITE_STATIC));

57 param = 0;

47 - rc = execute_insert(res);

58 + rc = sqlite3_step_monitored(res);

59 if (unlikely(rc != SQLITE_DONE)) {

60 error_report("HEALTH [%s]: Failed to update health log, rc = %d", rrdhost_hostname(host), rc);

61 }

63 done:

64 REPORT_BIND_FAIL(res, param);

54 - SQLITE_FINALIZE(res);

65 + if (is_health_thread)

66 + SQLITE_RESET(res);

67 + else

68 + SQLITE_FINALIZE(res);

69 }

71 /* Health related SQL queries

@@ -149,7 +163,20 @@ static void insert_alert_queue(

163 RRDCALC_STATUS old_status,

164 RRDCALC_STATUS new_status)

165 {

166 + static __thread sqlite3_stmt *compiled_res = NULL;

167 sqlite3_stmt *res = NULL;

168 +

169 + if (is_health_thread) {

170 + if (!compiled_res) {

171 + if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_INSERT_ALERT_PENDING_QUEUE, &compiled_res))

172 + return;

173 + }

174 + res = compiled_res;

175 + } else {

176 + if (!PREPARE_STATEMENT(db_meta, SQL_INSERT_ALERT_PENDING_QUEUE, &res))

177 + return;

178 + }

179 +

180 int rc;

181

182 if (!host->aclk_config)

@@ -169,14 +196,17 @@ static void insert_alert_queue(

196 SQLITE_BIND_FAIL(done, sqlite3_bind_int(res, ++param, submit_delay));

197

198 param = 0;

172 - rc = execute_insert(res);

199 + rc = sqlite3_step_monitored(res);

200 if (rc != SQLITE_DONE)

201 error_report(

202 "HEALTH [%s]: Failed to execute insert_alert_queue, rc = %d", rrdhost_hostname(host), rc);

203

204 done:

205 REPORT_BIND_FAIL(res, param);

179 - SQLITE_FINALIZE(res);

206 + if (is_health_thread)

207 + SQLITE_RESET(res);

208 + else

209 + SQLITE_FINALIZE(res);

210 }

211

212 #define SQL_INSERT_HEALTH_LOG_DETAIL \

@@ -189,11 +219,21 @@ done:

219

220 static void sql_health_alarm_log_insert_detail(RRDHOST *host, uint64_t health_log_id, ALARM_ENTRY *ae)

221 {

222 + static __thread sqlite3_stmt *compiled_res = NULL;

223 sqlite3_stmt *res = NULL;

193 - int rc;

224

195 - if (!PREPARE_STATEMENT(db_meta, SQL_INSERT_HEALTH_LOG_DETAIL, &res))

196 - return;

225 + if (is_health_thread) {

226 + if (!compiled_res) {

227 + if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_INSERT_HEALTH_LOG_DETAIL, &compiled_res))

228 + return;

229 + }

230 + res = compiled_res;

231 + } else {

232 + if (!PREPARE_STATEMENT(db_meta, SQL_INSERT_HEALTH_LOG_DETAIL, &res))

233 + return;

234 + }

235 +

236 + int rc;

237

238 int param = 0;

239 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, (sqlite3_int64)health_log_id));

@@ -221,7 +261,7 @@ static void sql_health_alarm_log_insert_detail(RRDHOST *host, uint64_t health_lo

261 SQLITE_BIND_FAIL(done, SQLITE3_BIND_STRING_OR_NULL(res, ++param, ae->summary));

262

263 param = 0;

224 - rc = execute_insert(res);

264 + rc = sqlite3_step_monitored(res);

265 if (rc == SQLITE_DONE)

266 ae->flags |= HEALTH_ENTRY_FLAG_SAVED;

267 else

@@ -230,7 +270,10 @@ static void sql_health_alarm_log_insert_detail(RRDHOST *host, uint64_t health_lo

270

271 done:

272 REPORT_BIND_FAIL(res, param);

233 - SQLITE_FINALIZE(res);

273 + if (is_health_thread)

274 + SQLITE_RESET(res);

275 + else

276 + SQLITE_FINALIZE(res);

277 }

278

279 #define SQL_INSERT_HEALTH_LOG \

@@ -243,12 +286,21 @@ done:

286

287 static void sql_health_alarm_log_insert(RRDHOST *host, ALARM_ENTRY *ae)

288 {

289 + static __thread sqlite3_stmt *compiled_res = NULL;

290 sqlite3_stmt *res = NULL;

291 int rc;

292 uint64_t health_log_id;

293

250 - if (!PREPARE_STATEMENT(db_meta, SQL_INSERT_HEALTH_LOG, &res))

251 - return;

294 + if (is_health_thread) {

295 + if (!compiled_res) {

296 + if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_INSERT_HEALTH_LOG, &compiled_res))

297 + return;

298 + }

299 + res = compiled_res;

300 + } else {

301 + if (!PREPARE_STATEMENT(db_meta, SQL_INSERT_HEALTH_LOG, &res))

302 + return;

303 + }

304

305 int param = 0;

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

@@ -275,7 +327,10 @@ static void sql_health_alarm_log_insert(RRDHOST *host, ALARM_ENTRY *ae)

327

328 done:

329 REPORT_BIND_FAIL(res, param);

278 - SQLITE_FINALIZE(res);

330 + if (is_health_thread)

331 + SQLITE_RESET(res);

332 + else

333 + SQLITE_FINALIZE(res);

334 }

335

336 void sql_health_alarm_log_save(RRDHOST *host, ALARM_ENTRY *ae)

@@ -340,7 +395,7 @@ bool sql_update_transition_in_health_log(RRDHOST *host, uint32_t alarm_id, nd_uu

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

396

397 param = 0;

343 - rc = execute_insert(res);

398 + rc = sqlite3_step_monitored(res);

399 if (unlikely(rc != SQLITE_DONE))

400 error_report("HEALTH [N/A]: Failed to execute SQL_INJECT_REMOVED_UPDATE_DETAIL, rc = %d", rc);

401

@@ -370,7 +425,7 @@ bool sql_set_updated_by_in_health_log_detail(uint32_t unique_id, uint32_t max_un

425 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, prev_transition_id, sizeof(*prev_transition_id), SQLITE_STATIC));

426

427 param = 0;

373 - rc = execute_insert(res);

428 + rc = sqlite3_step_monitored(res);

429 if (unlikely(rc != SQLITE_DONE))

430 error_report("HEALTH [N/A]: Failed to execute SQL_INJECT_REMOVED_UPDATE_DETAIL, rc = %d", rc);

431

@@ -418,7 +473,6 @@ static void sql_inject_removed_status(

473 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, last_transition, sizeof(*last_transition), SQLITE_STATIC));

474

475 param = 0;

421 - //int rc = execute_insert(res);

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

477 //update the old entry in health_log_detail

478 sql_set_updated_by_in_health_log_detail(unique_id, max_unique_id, last_transition);

@@ -521,7 +575,7 @@ static void sql_remove_alerts_from_deleted_charts(RRDHOST *host, nd_uuid_t *host

575 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, actual_uuid, sizeof(*actual_uuid), SQLITE_STATIC));

576

577 param = 0;

524 - ret = execute_insert(res);

578 + ret = sqlite3_step_monitored(res);

579 if (ret != SQLITE_DONE)

580 error_report("Failed to execute command to delete missing charts from health_log");

581

@@ -868,7 +922,7 @@ void sql_alert_store_config(RRD_ALERT_PROTOTYPE *ap)

922 SQLITE_BIND_FAIL(done, sqlite3_bind_int(res, ++param, ap->config.data_source));

923

924 param = 0;

871 - int rc = execute_insert(res);

925 + int rc = sqlite3_step_monitored(res);

926 if (unlikely(rc != SQLITE_DONE))

927 error_report("Failed to store alert config, rc = %d", rc);

928

src/health/health_event_loop.c

@@ -643,8 +643,10 @@ static void health_event_loop_for_host(RRDHOST *host, bool apply_hibernation_del

643 }

644 }

645

646 +__thread bool is_health_thread = false;

647 static void health_event_loop(void) {

648

649 + is_health_thread = true;

650 while(service_running(SERVICE_HEALTH)) {

651 if(!stream_control_health_should_be_running()) {

652 worker_is_idle();