@cryptotaxi247 / netdata-1 / commits / a706491f7

Improve agent to cloud synchronization performance (#12348)

* Switch to prepare statement when storing active charts / dimensions * Switch to prepare statement when storing chart labels * Switch to prepare statement when doing a node id lookup * Switch to prepare statement when loading the node id for a host * Improve performance by avoiding db query * Use prepare statement when counting pending chart messages to send to the cloud * Delay locking while preparing commands * No need to use buffer, avoid memory allocation overhead * Switch to prepare statement when loading pending chart updates to send to the cloud

Stelios Fragkakis committed Mar 9, 2022 at 19:54 UTC a706491f775fbd959fc97e38cf902f6748855e18

3 files changed +93 -95

aclk/aclk_query.c

@@ -58,6 +58,13 @@ static RRDHOST *node_id_2_rrdhost(const char *node_id)

58 {

59 int res;

60 uuid_t node_id_bin, host_id_bin;

61 +

62 + rrd_rdlock();

63 + RRDHOST *host = find_host_by_node_id((char *) node_id);

64 + rrd_unlock();

65 + if (host)

66 + return host;

67 +

68 char host_id[UUID_STR_LEN];

69 if (uuid_parse(node_id, node_id_bin)) {

70 error("Couldn't parse UUID %s", node_id);

database/sqlite/sqlite_aclk_chart.c

+53 -68

@@ -29,12 +29,11 @@ static int payload_sent(char *uuid_str, uuid_t *uuid, void *payload, size_t payl

29 int send_status = 0;

31 if (unlikely(!res)) {

32 - BUFFER *sql = buffer_create(1024);

33 - buffer_sprintf(sql,"SELECT 1 FROM aclk_chart_latest_%s acl, aclk_chart_payload_%s acp "

32 + char sql[ACLK_SYNC_QUERY_SIZE];

33 + snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "SELECT 1 FROM aclk_chart_latest_%s acl, aclk_chart_payload_%s acp "

34 "WHERE acl.unique_id = acp.unique_id AND acl.uuid = @uuid AND acp.payload = @payload;",

35 uuid_str, uuid_str);

36 - rc = prepare_statement(db_meta, (char *) buffer_tostring(sql), &res);

37 - buffer_free(sql);

36 + rc = prepare_statement(db_meta, sql, &res);

37 if (rc != SQLITE_OK) {

38 error_report("Failed to prepare statement to check payload data");

39 return 0;

@@ -70,14 +69,11 @@ static int aclk_add_chart_payload(struct aclk_database_worker_config *wc, uuid_t

69 return 0;

71 if (unlikely(!res_chart)) {

73 - BUFFER *sql = buffer_create(1024);

74 -

75 - buffer_sprintf(sql,"INSERT INTO aclk_chart_payload_%s (unique_id, uuid, claim_id, date_created, type, payload) " \

76 - "VALUES (@unique_id, @uuid, @claim_id, strftime('%%s','now'), @type, @payload);", wc->uuid_str);

77 -

78 - rc = prepare_statement(db_meta, (char *) buffer_tostring(sql), &res_chart);

79 - buffer_free(sql);

80 -

72 + char sql[ACLK_SYNC_QUERY_SIZE];

73 + snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1,

74 + "INSERT INTO aclk_chart_payload_%s (unique_id, uuid, claim_id, date_created, type, payload) " \

75 + "VALUES (@unique_id, @uuid, @claim_id, strftime('%%s','now'), @type, @payload);", wc->uuid_str);

76 + rc = prepare_statement(db_meta, sql, &res_chart);

77 if (rc != SQLITE_OK) {

78 error_report("Failed to prepare statement to store chart payload data");

79 return 1;

@@ -320,21 +316,20 @@ void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_d

316 uint64_t last_sequence;

317 time_t last_timestamp = 0;

318

323 - BUFFER *sql = buffer_create(1024);

324 -

325 - sqlite3_stmt *res = NULL;

326 -

327 - buffer_sprintf(sql, "SELECT ac.sequence_id, acp.payload, ac.date_created, ac.type, ac.uuid " \

328 - "FROM aclk_chart_%s ac, aclk_chart_payload_%s acp " \

329 - "WHERE ac.date_submitted IS NULL AND ac.unique_id = acp.unique_id AND ac.update_count > 0 " \

330 - "AND acp.claim_id = @claim_id ORDER BY ac.sequence_id ASC LIMIT %d;", wc->uuid_str, wc->uuid_str, limit);

319 + char sql[ACLK_SYNC_QUERY_SIZE];

320 + static __thread sqlite3_stmt *res = NULL;

321

332 - rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);

333 - if (rc != SQLITE_OK) {

334 - error_report("Failed to prepare statement when trying to send a chart update via ACLK");

335 - buffer_free(sql);

336 - freez(claim_id);

337 - return;

322 + if (unlikely(!res)) {

323 + snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1,"SELECT ac.sequence_id, acp.payload, ac.date_created, ac.type, ac.uuid " \

324 + "FROM aclk_chart_%s ac, aclk_chart_payload_%s acp " \

325 + "WHERE ac.date_submitted IS NULL AND ac.unique_id = acp.unique_id AND ac.update_count > 0 " \

326 + "AND acp.claim_id = @claim_id ORDER BY ac.sequence_id ASC LIMIT %d;", wc->uuid_str, wc->uuid_str, limit);

327 + rc = prepare_statement(db_meta, sql, &res);

328 + if (rc != SQLITE_OK) {

329 + error_report("Failed to prepare statement when trying to send a chart update via ACLK");

330 + freez(claim_id);

331 + return;

332 + }

333 }

334

335 rc = sqlite3_bind_blob(res, 1, claim_uuid , sizeof(claim_uuid), SQLITE_STATIC);

@@ -388,21 +383,18 @@ void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_d

383 error_report("Failed to reset statement when pushing chart events, rc = %d", rc);

384

385 if (likely(first_sequence)) {

391 - buffer_flush(sql);

386

387 db_lock();

394 - buffer_sprintf(sql, "UPDATE aclk_chart_%s SET status = NULL, date_submitted=strftime('%%s','now') "

388 + snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "UPDATE aclk_chart_%s SET status = NULL, date_submitted=strftime('%%s','now') "

389 "WHERE date_submitted IS NULL AND sequence_id BETWEEN %" PRIu64 " AND %" PRIu64 ";",

390 wc->uuid_str, first_sequence, last_sequence);

397 - db_execute(buffer_tostring(sql));

398 -

399 - buffer_flush(sql);

400 - buffer_sprintf(sql, "INSERT OR REPLACE INTO aclk_chart_latest_%s (uuid, unique_id, date_submitted) "

391 + db_execute(sql);

392 + snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "INSERT OR REPLACE INTO aclk_chart_latest_%s (uuid, unique_id, date_submitted) "

393 " SELECT uuid, unique_id, date_submitted FROM aclk_chart_%s s "

394 " WHERE date_submitted IS NOT NULL AND sequence_id BETWEEN %" PRIu64 " AND %" PRIu64

395 " ;",

396 wc->uuid_str, wc->uuid_str, first_sequence, last_sequence);

405 - db_execute(buffer_tostring(sql));

397 + db_execute(sql);

398 db_unlock();

399

400 aclk_chart_inst_and_dim_update(payload_list, payload_list_size, is_dim, position_list, wc->batch_id);

@@ -436,11 +428,10 @@ void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_d

428 freez(is_dim);

429

430 bind_fail:

439 - rc = sqlite3_finalize(res);

431 + rc = sqlite3_reset(res);

432 if (unlikely(rc != SQLITE_OK))

441 - error_report("Failed to finalize statement when pushing chart events, rc = %d", rc);

433 + error_report("Failed to reset statement when pushing chart events, rc = %d", rc);

434

443 - buffer_free(sql);

435 freez(claim_id);

436 return;

437 }

@@ -521,15 +512,15 @@ void aclk_receive_chart_ack(struct aclk_database_worker_config *wc, struct aclk_

512

513 log_access("IN [%s (%s)]: Received ack chart sequence id %"PRIu64, wc->node_id, wc->host ? wc->host->hostname : "N/A", cmd.param1);

514

524 - BUFFER *sql = buffer_create(1024);

515 + char sql[ACLK_SYNC_QUERY_SIZE];

516

526 - buffer_sprintf(sql, "UPDATE aclk_chart_%s SET date_updated=strftime('%%s','now') WHERE sequence_id <= @sequence_id "

527 - "AND date_submitted IS NOT NULL AND date_updated IS NULL;", wc->uuid_str);

517 + snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1,"UPDATE aclk_chart_%s SET date_updated=strftime('%%s','now') WHERE sequence_id <= @sequence_id "

518 + "AND date_submitted IS NOT NULL AND date_updated IS NULL;", wc->uuid_str);

519

529 - rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);

520 + rc = sqlite3_prepare_v2(db_meta, sql, -1, &res, 0);

521 if (rc != SQLITE_OK) {

531 - error_report("Failed to prepare statement count sequence ids in the database");

532 - goto prepare_fail;

522 + error_report("Failed to prepare statement to ack chart sequence ids");

523 + return;

524 }

525

526 rc = sqlite3_bind_int64(res, 1, (uint64_t) cmd.param1);

@@ -540,13 +531,10 @@ void aclk_receive_chart_ack(struct aclk_database_worker_config *wc, struct aclk_

531 if (rc != SQLITE_DONE)

532 error_report("Failed to ACK sequence id, rc = %d", rc);

533

543 - bind_fail:

544 - if (unlikely(sqlite3_finalize(res) != SQLITE_OK))

545 - error_report("Failed to finalize statement to ACK older sequence ids, rc = %d", rc);

546 -

547 - prepare_fail:

548 - buffer_free(sql);

549 - return;

534 +bind_fail:

535 + if (unlikely(sqlite3_finalize(res) != SQLITE_OK))

536 + error_report("Failed to finalize statement to ACK older sequence ids, rc = %d", rc);

537 + return;

538 }

539

540 void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)

@@ -556,12 +544,11 @@ void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct acl

544 wc->uuid_str, cmd.param1);

545 db_execute(buffer_tostring(sql));

546 if (cmd.param1 == 1) {

559 - db_lock();

547 buffer_flush(sql);

548 log_access("IN [%s (%s)]: Received chart full resync.", wc->node_id, wc->host ? wc->host->hostname : "N/A");

549 buffer_sprintf(sql, "DELETE FROM aclk_chart_payload_%s; DELETE FROM aclk_chart_%s; " \

550 "DELETE FROM aclk_chart_latest_%s;", wc->uuid_str, wc->uuid_str, wc->uuid_str);

564 -

551 + db_lock();

552 db_execute("BEGIN TRANSACTION;");

553 db_execute(buffer_tostring(sql));

554 db_execute("COMMIT TRANSACTION;");

@@ -909,43 +896,43 @@ failed:

896

897 uint32_t sql_get_pending_count(struct aclk_database_worker_config *wc)

898 {

912 - BUFFER *sql = buffer_create(1024);

913 - sqlite3_stmt *res = NULL;

899 + char sql[ACLK_SYNC_QUERY_SIZE];

900 + static __thread sqlite3_stmt *res = NULL;

901

915 - buffer_sprintf(sql,"SELECT count(1) FROM aclk_chart_%s ac WHERE ac.date_submitted IS NULL;", wc->uuid_str);

902 + snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "SELECT count(1) FROM aclk_chart_%s ac WHERE ac.date_submitted IS NULL;", wc->uuid_str);

903

904 int rc;

905 uint32_t chart_payload_count = 0;

919 - rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);

920 - if (rc != SQLITE_OK) {

921 - error_report("Failed to prepare statement to count pending messages");

922 - goto fail;

906 + if (unlikely(!res)) {

907 + rc = prepare_statement(db_meta, sql, &res);

908 + if (rc != SQLITE_OK) {

909 + error_report("Failed to prepare statement to count pending messages");

910 + return 0;

911 + }

912 }

913 while (sqlite3_step(res) == SQLITE_ROW)

914 chart_payload_count = (uint32_t) sqlite3_column_int(res, 0);

915

927 - rc = sqlite3_finalize(res);

916 + rc = sqlite3_reset(res);

917 if (unlikely(rc != SQLITE_OK))

918 error_report("Failed to reset statement when fetching pending messages, rc = %d", rc);

919

931 -fail:

932 - buffer_free(sql);

920 return chart_payload_count;

921 }

922

923 void sql_get_last_chart_sequence(struct aclk_database_worker_config *wc)

924 {

938 - BUFFER *sql = buffer_create(1024);

925 + char sql[ACLK_SYNC_QUERY_SIZE];

926

940 - buffer_sprintf(sql,"SELECT ac.sequence_id, ac.date_created FROM aclk_chart_%s ac " \

941 - "WHERE ac.date_submitted IS NOT NULL ORDER BY ac.sequence_id DESC LIMIT 1;", wc->uuid_str);

927 + snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "SELECT ac.sequence_id, ac.date_created FROM aclk_chart_%s ac " \

928 + "WHERE ac.date_submitted IS NOT NULL ORDER BY ac.sequence_id DESC LIMIT 1;", wc->uuid_str);

929

930 int rc;

931 sqlite3_stmt *res = NULL;

945 - rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);

932 + rc = sqlite3_prepare_v2(db_meta, sql, -1, &res, 0);

933 if (rc != SQLITE_OK) {

934 error_report("Failed to prepare statement to find last chart sequence id");

948 - goto fail;

935 + return;

936 }

937

938 wc->chart_sequence_id = 0;

@@ -961,8 +948,6 @@ void sql_get_last_chart_sequence(struct aclk_database_worker_config *wc)

948 if (unlikely(rc != SQLITE_OK))

949 error_report("Failed to reset statement when fetching chart sequence info, rc = %d", rc);

950

964 -fail:

965 - buffer_free(sql);

951 return;

952 }

953

database/sqlite/sqlite_functions.c

+33 -27

@@ -121,7 +121,7 @@ static int store_active_uuid_object(sqlite3_stmt **res, char *statement, uuid_t

121

122 // Check if we should need to prepare the statement

123 if (!*res) {

124 - rc = sqlite3_prepare_v2(db_meta, statement, -1, res, 0);

124 + rc = prepare_statement(db_meta, statement, res);

125 if (unlikely(rc != SQLITE_OK)) {

126 error_report("Failed to prepare statement to store active object, rc = %d", rc);

127 return rc;

@@ -142,7 +142,7 @@ static int store_active_uuid_object(sqlite3_stmt **res, char *statement, uuid_t

142 */

143 void store_active_chart(uuid_t *chart_uuid)

144 {

145 - sqlite3_stmt *res = NULL;

145 + static __thread sqlite3_stmt *res = NULL;

146 int rc;

147

148 if (unlikely(!db_meta)) {

@@ -158,7 +158,7 @@ void store_active_chart(uuid_t *chart_uuid)

158 if (rc != SQLITE_DONE)

159 error_report("Failed to store active chart, rc = %d", rc);

160

161 - rc = sqlite3_finalize(res);

161 + rc = sqlite3_reset(res);

162 if (unlikely(rc != SQLITE_OK))

163 error_report("Failed to finalize statement in store active chart, rc = %d", rc);

164 return;

@@ -170,7 +170,7 @@ void store_active_chart(uuid_t *chart_uuid)

170 */

171 void store_active_dimension(uuid_t *dimension_uuid)

172 {

173 - sqlite3_stmt *res = NULL;

173 + static __thread sqlite3_stmt *res = NULL;

174 int rc;

175

176 if (unlikely(!db_meta)) {

@@ -186,7 +186,7 @@ void store_active_dimension(uuid_t *dimension_uuid)

186 if (rc != SQLITE_DONE)

187 error_report("Failed to store active dimension, rc = %d", rc);

188

189 - rc = sqlite3_finalize(res);

189 + rc = sqlite3_reset(res);

190 if (unlikely(rc != SQLITE_OK))

191 error_report("Failed to finalize statement in store active dimension, rc = %d", rc);

192 return;

@@ -1335,7 +1335,7 @@ void add_migrated_file(char *path, uint64_t file_size)

1335

1336 void sql_store_chart_label(uuid_t *chart_uuid, int source_type, char *label, char *value)

1337 {

1338 - sqlite3_stmt *res = NULL;

1338 + static __thread sqlite3_stmt *res = NULL;

1339 int rc;

1340

1341 if (unlikely(!db_meta)) {

@@ -1344,10 +1344,12 @@ void sql_store_chart_label(uuid_t *chart_uuid, int source_type, char *label, cha

1344 return;

1345 }

1346

1347 - rc = sqlite3_prepare_v2(db_meta, SQL_INS_CHART_LABEL, -1, &res, 0);

1348 - if (unlikely(rc != SQLITE_OK)) {

1349 - error_report("Failed to prepare statement store chart labels");

1350 - return;

1347 + if (unlikely(!res)) {

1348 + rc = prepare_statement(db_meta, SQL_INS_CHART_LABEL, &res);

1349 + if (unlikely(rc != SQLITE_OK)) {

1350 + error_report("Failed to prepare statement store chart labels");

1351 + return;

1352 + }

1353 }

1354

1355 rc = sqlite3_bind_blob(res, 1, chart_uuid, sizeof(*chart_uuid), SQLITE_STATIC);

@@ -1379,8 +1381,8 @@ void sql_store_chart_label(uuid_t *chart_uuid, int source_type, char *label, cha

1381 error_report("Failed to store chart label entry, rc = %d", rc);

1382

1383 failed:

1382 - if (unlikely(sqlite3_finalize(res) != SQLITE_OK))

1383 - error_report("Failed to finalize the prepared statement when storing chart label information");

1384 + if (unlikely(sqlite3_reset(res) != SQLITE_OK))

1385 + error_report("Failed to reset the prepared statement when storing chart label information");

1386

1387 return;

1388 }

@@ -1881,7 +1883,7 @@ failed:

1883

1884 int get_host_id(uuid_t *node_id, uuid_t *host_id)

1885 {

1884 - sqlite3_stmt *res = NULL;

1886 + static __thread sqlite3_stmt *res = NULL;

1887 int rc;

1888

1889 if (unlikely(!db_meta)) {

@@ -1890,10 +1892,12 @@ int get_host_id(uuid_t *node_id, uuid_t *host_id)

1892 return 1;

1893 }

1894

1893 - rc = sqlite3_prepare_v2(db_meta, SQL_SELECT_HOST_BY_NODE_ID, -1, &res, 0);

1894 - if (unlikely(rc != SQLITE_OK)) {

1895 - error_report("Failed to prepare statement to select node instance information for a node");

1896 - return 1;

1895 + if (unlikely(!res)) {

1896 + rc = prepare_statement(db_meta, SQL_SELECT_HOST_BY_NODE_ID, &res);

1897 + if (unlikely(rc != SQLITE_OK)) {

1898 + error_report("Failed to prepare statement to select node instance information for a node");

1899 + return 1;

1900 + }

1901 }

1902

1903 rc = sqlite3_bind_blob(res, 1, node_id, sizeof(*node_id), SQLITE_STATIC);

@@ -1907,8 +1911,8 @@ int get_host_id(uuid_t *node_id, uuid_t *host_id)

1911 uuid_copy(*host_id, *((uuid_t *) sqlite3_column_blob(res, 0)));

1912

1913 failed:

1910 - if (unlikely(sqlite3_finalize(res) != SQLITE_OK))

1911 - error_report("Failed to finalize the prepared statement when selecting node instance information");

1914 + if (unlikely(sqlite3_reset(res) != SQLITE_OK))

1915 + error_report("Failed to reset the prepared statement when selecting node instance information");

1916

1917 return (rc == SQLITE_ROW) ? 0 : -1;

1918 }

@@ -2060,7 +2064,7 @@ failed:

2064

2065 void sql_load_node_id(RRDHOST *host)

2066 {

2063 - sqlite3_stmt *res = NULL;

2067 + static __thread sqlite3_stmt *res = NULL;

2068 int rc;

2069

2070 if (unlikely(!db_meta)) {

@@ -2069,11 +2073,13 @@ void sql_load_node_id(RRDHOST *host)

2073 return;

2074 }

2075

2072 - rc = sqlite3_prepare_v2(db_meta, SQL_GET_HOST_NODE_ID, -1, &res, 0);

2073 - if (unlikely(rc != SQLITE_OK)) {

2074 - error_report("Failed to prepare statement to fetch node id");

2075 - return;

2076 - };

2076 + if (unlikely(!res)) {

2077 + rc = prepare_statement(db_meta, SQL_GET_HOST_NODE_ID, &res);

2078 + if (unlikely(rc != SQLITE_OK)) {

2079 + error_report("Failed to prepare statement to fetch node id");

2080 + return;

2081 + };

2082 + }

2083

2084 rc = sqlite3_bind_blob(res, 1, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC);

2085 if (unlikely(rc != SQLITE_OK)) {

@@ -2090,8 +2096,8 @@ void sql_load_node_id(RRDHOST *host)

2096 }

2097

2098 failed:

2093 - if (unlikely(sqlite3_finalize(res) != SQLITE_OK))

2094 - error_report("Failed to finalize the prepared statement when loading node instance information");

2099 + if (unlikely(sqlite3_reset(res) != SQLITE_OK))

2100 + error_report("Failed to reset the prepared statement when loading node instance information");

2101

2102 return;

2103 };