@cryptotaxi247 / netdata-1 / commits / e9efad18e

Improve the ACLK sync process for the new cloud architecture (#11744)

* Move retention code to the charts * Log information about node registration and updates * Prevent deadlock if aclk_database_enq_cmd locks for a node * Improve message (indicate that it comes from alerts). This will be improved in a followup PR * Disable parts that can't be used if the new cloud env is not available * Set dimension FLAG if message has been queued * Queue messages using the correct protocol enabled * Cleanup unused functions Rename functions that queue charts and dimensions Improve the generic chart payload add function Add a counter for pending charts/dimension payloads to avoid polling the db Delay the retention update message until we are done with the updates Fix full resync command to handle sequence_id = 0 correctly Disable functions not needed when the new cloud env functionality is not compiled * Add chart_payload count and retry count Output information or error message if we fail to queue chart/dimension PUSH commands Only try to queue commands if we have chart_payload_count>0 Remove the event loop shutdown opcode handle * Improve detection of shutdown (check netdata_exit) * Adjusting info messages

Stelios Fragkakis committed Nov 3, 2021 at 19:18 UTC e9efad18e8c53ac4cb777d553efe4b22cc874b7b
8 files changed +414 -415
aclk/aclk.c
+8
@@ -978,6 +978,7 @@ void ng_aclk_host_state_update(RRDHOST *host, int cmd)
978 create_query->data.node_creation.hops = (uint32_t) host->system_info->hops;
979 create_query->data.node_creation.hostname = strdupz(host->hostname);
980 create_query->data.node_creation.machine_guid = strdupz(host->machine_guid);
981 + info("Registering host=%s, hops=%u",host->machine_guid, host->system_info->hops);
982 aclk_queue_query(create_query);
983 return;
984 }
@@ -992,6 +993,8 @@ void ng_aclk_host_state_update(RRDHOST *host, int cmd)
993 uuid_unparse_lower(node_id, (char*)query->data.node_update.node_id);
994 query->data.node_update.queryable = 1;
995 query->data.node_update.session_id = aclk_session_newarch;
996 + info("Queuing status update for node=%s, live=%d, hops=%u",(char*)query->data.node_update.node_id, cmd,
997 + host->system_info->hops);
998 aclk_queue_query(query);
999 }
1000
@@ -1015,6 +1018,9 @@ void aclk_send_node_instances()
1018 uuid_unparse_lower(list->node_id, (char*)query->data.node_update.node_id);
1019 query->data.node_update.queryable = 1;
1020 query->data.node_update.session_id = aclk_session_newarch;
1021 + info("Queuing status update for node=%s, live=%d, hops=%d",(char*)query->data.node_update.node_id,
1022 + list->live,
1023 + list->hops);
1024 aclk_queue_query(query);
1025 } else {
1026 aclk_query_t create_query;
@@ -1026,6 +1032,8 @@ void aclk_send_node_instances()
1032 create_query->data.node_creation.hostname = list->hostname;
1033 create_query->data.node_creation.machine_guid = mallocz(UUID_STR_LEN);
1034 uuid_unparse_lower(list->host_id, (char*)create_query->data.node_creation.machine_guid);
1035 + info("Queuing registration for host=%s, hops=%d",(char*)create_query->data.node_creation.machine_guid,
1036 + list->hops);
1037 aclk_queue_query(create_query);
1038 }
1039
database/rrdset.c
+4 -10
@@ -1391,19 +1391,12 @@ void rrdset_done(RRDSET *st) {
1391 rrdset_rdlock(st);
1392
1393 #ifdef ENABLE_ACLK
1394 - #ifdef ENABLE_NEW_CLOUD_PROTOCOL
1394 if (unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
1395 if (st->counter_done >= RRDSET_MINIMUM_LIVE_COUNT) {
1397 - if (likely(!sql_queue_chart_to_aclk(st)))
1396 + if (likely(!queue_chart_to_aclk(st)))
1397 rrdset_flag_set(st, RRDSET_FLAG_ACLK);
1398 }
1399 }
1401 - #else
1402 - if (unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
1403 - rrdset_flag_set(st, RRDSET_FLAG_ACLK);
1404 - aclk_update_chart(st->rrdhost, st->id, 1);
1405 - }
1406 - #endif
1400 #endif
1401
1402 if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE))) {
@@ -1813,12 +1806,13 @@ after_second_database_work:
1806 if (rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED))
1807 continue;
1808
1816 -#ifdef ENABLE_NEW_CLOUD_PROTOCOL
1809 +#if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
1810 int live = ((mark - rd->last_collected_time.tv_sec) < (RRDSET_MINIMUM_LIVE_COUNT * rd->update_every));
1811 if (unlikely(live != rd->state->aclk_live_status)) {
1812 if (likely(rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
1820 - if (likely(!sql_queue_dimension_to_aclk(rd))) {
1813 + if (likely(!queue_dimension_to_aclk(rd))) {
1814 rd->state->aclk_live_status = live;
1815 + rrddim_flag_set(rd, RRDDIM_FLAG_ACLK);
1816 }
1817 }
1818 }
database/sqlite/sqlite_aclk.c
+42 -24
@@ -165,9 +165,9 @@ int aclk_worker_enq_cmd(char *node_id, struct aclk_database_cmd *cmd)
165 break;
166 wc = wc->next;
167 }
168 + uv_mutex_unlock(&aclk_async_lock);
169 if (wc)
170 aclk_database_enq_cmd(wc, cmd);
170 - uv_mutex_unlock(&aclk_async_lock);
171 return (wc == NULL);
172 }
173
@@ -261,12 +261,22 @@ static void timer_cb(uv_timer_t* handle)
261 wc->rotation_after += ACLK_DATABASE_ROTATION_INTERVAL;
262 }
263
264 - if (wc->chart_updates && !wc->chart_pending) {
264 + if (wc->chart_updates && !wc->chart_pending && wc->chart_payload_count) {
265 cmd.opcode = ACLK_DATABASE_PUSH_CHART;
266 cmd.count = ACLK_MAX_CHART_BATCH;
267 cmd.param1 = ACLK_MAX_CHART_BATCH_COUNT;
268 - if (!aclk_database_enq_cmd_noblock(wc, &cmd))
268 + if (!aclk_database_enq_cmd_noblock(wc, &cmd)) {
269 + if (wc->retry_count)
270 + info("Queued chart/dimension payload command %s, retry count = %u", wc->host_guid, wc->retry_count);
271 wc->chart_pending = 1;
272 + wc->retry_count = 0;
273 + } else {
274 + wc->retry_count++;
275 + if (wc->retry_count % 100 == 0)
276 + error_report("Failed to queue chart/dimension payload command %s, retry count = %u",
277 + wc->host_guid,
278 + wc->retry_count);
279 + }
280 }
281
282 if (wc->alert_updates) {
@@ -284,7 +294,7 @@ void aclk_database_worker(void *arg)
294 {
295 struct aclk_database_worker_config *wc = arg;
296 uv_loop_t *loop;
287 - int shutdown, ret;
297 + int ret;
298 enum aclk_database_opcode opcode;
299 uv_timer_t timer_req;
300 struct aclk_database_cmd cmd;
@@ -324,25 +334,27 @@ void aclk_database_worker(void *arg)
334 timer_req.data = wc;
335 fatal_assert(0 == uv_timer_start(&timer_req, timer_cb, TIMER_PERIOD_MS, TIMER_PERIOD_MS));
336
327 - wc->error = 0;
328 - shutdown = 0;
329 -
337 + wc->retry_count = 0;
338 wc->node_info_send = (wc->host && !localhost);
339 aclk_add_worker_thread(wc);
340 info("Starting ACLK sync thread for host %s -- scratch area %lu bytes", wc->host_guid, sizeof(*wc));
341
342 memset(&cmd, 0, sizeof(cmd));
335 - sql_get_last_chart_sequence(wc, cmd);
343 +#ifdef ENABLE_NEW_CLOUD_PROTOCOL
344 + sql_get_last_chart_sequence(wc);
345 + wc->chart_payload_count = sql_get_pending_count(wc);
346 + if (!wc->chart_payload_count)
347 + info("%s: No pending charts and dimensions detected during startup", wc->host_guid);
348 +#endif
349 wc->chart_updates = 0;
337 - wc->alert_updates = 0;
350 wc->startup_time = now_realtime_sec();
351 wc->cleanup_after = wc->startup_time + ACLK_DATABASE_CLEANUP_FIRST;
352 wc->rotation_after = wc->startup_time + ACLK_DATABASE_ROTATION_DELAY;
341 - while (likely(shutdown == 0)) {
342 - uv_run(loop, UV_RUN_DEFAULT);
353 + wc->alert_updates = 0;
354
344 - if (netdata_exit)
345 - shutdown = 1;
355 + debug(D_ACLK_SYNC,"Node %s reports pending message count = %u", wc->node_id, wc->chart_payload_count);
356 + while (likely(!netdata_exit)) {
357 + uv_run(loop, UV_RUN_DEFAULT);
358
359 /* wait for commands */
360 cmd_batch_size = 0;
@@ -350,6 +362,10 @@ void aclk_database_worker(void *arg)
362 if (unlikely(cmd_batch_size >= MAX_CMD_BATCH_SIZE))
363 break;
364 cmd = aclk_database_deq_cmd(wc);
365 +
366 + if (netdata_exit)
367 + break;
368 +
369 opcode = cmd.opcode;
370 ++cmd_batch_size;
371 switch (opcode) {
@@ -370,6 +386,7 @@ void aclk_database_worker(void *arg)
386 break;
387
388 // CHART / DIMENSION OPERATIONS
389 +#ifdef ENABLE_NEW_CLOUD_PROTOCOL
390 case ACLK_DATABASE_ADD_CHART:
391 debug(D_ACLK_SYNC, "Adding chart event for %s", wc->host_guid);
392 aclk_add_chart_event(wc, cmd);
@@ -394,7 +411,7 @@ void aclk_database_worker(void *arg)
411 debug(D_ACLK_SYNC, "RESET chart SEQ for %s to %"PRIu64, wc->uuid_str, (uint64_t) cmd.param1);
412 aclk_receive_chart_reset(wc, cmd);
413 break;
397 -
414 +#endif
415 // ALERTS
416 case ACLK_DATABASE_ADD_ALERT:
417 debug(D_ACLK_SYNC,"Adding alert event for %s", wc->host_guid);
@@ -426,10 +443,15 @@ void aclk_database_worker(void *arg)
443 debug(D_ACLK_SYNC,"Sending node info for %s", wc->uuid_str);
444 sql_build_node_info(wc, cmd);
445 break;
446 +#ifdef ENABLE_NEW_CLOUD_PROTOCOL
447 + case ACLK_DATABASE_DIM_DELETION:
448 + debug(D_ACLK_SYNC,"Sending dimension deletion information %s", wc->uuid_str);
449 + break;
450 case ACLK_DATABASE_UPD_RETENTION:
451 debug(D_ACLK_SYNC,"Sending retention info for %s", wc->uuid_str);
452 aclk_update_retention(wc, cmd);
453 break;
454 +#endif
455
456 // NODE_INSTANCE DETECTION
457 case ACLK_DATABASE_TIMER:
@@ -437,7 +459,7 @@ void aclk_database_worker(void *arg)
459 if (claimed()) {
460 wc->host = rrdhost_find_by_guid(wc->host_guid, 0);
461 if (wc->host) {
440 - info("HOST %s detected as active and claimed !!!", wc->host->hostname);
462 + info("HOST %s (%s) detected as active", wc->host->hostname, wc->host_guid);
463 snprintfz(threadname, NETDATA_THREAD_NAME_MAX, "AS_%s", wc->host->hostname);
464 uv_thread_set_name_np(wc->thread, threadname);
465 wc->host->dbsync_worker = wc;
@@ -452,11 +474,6 @@ void aclk_database_worker(void *arg)
474 wc->node_info_send = aclk_database_enq_cmd_noblock(wc, &cmd);
475 }
476 break;
455 - case ACLK_DATABASE_SHUTDOWN:
456 - shutdown = 1;
457 - fatal_assert(0 == uv_timer_stop(&timer_req));
458 - uv_close((uv_handle_t *)&timer_req, NULL);
459 - break;
477 default:
478 debug(D_ACLK_SYNC, "%s: default.", __func__);
479 break;
@@ -466,8 +483,11 @@ void aclk_database_worker(void *arg)
483 } while (opcode != ACLK_DATABASE_NOOP);
484 }
485
486 + if (!uv_timer_stop(&timer_req))
487 + uv_close((uv_handle_t *)&timer_req, NULL);
488 +
489 /* cleanup operations of the event loop */
470 - info("Shutting down ACLK_DATABASE engine event loop.");
490 + info("Shutting down ACLK sync event loop.");
491
492 /*
493 * uv_async_send after uv_close does not seem to crash in linux at the moment,
@@ -477,7 +497,7 @@ void aclk_database_worker(void *arg)
497 uv_close((uv_handle_t *)&wc->async, NULL);
498 uv_run(loop, UV_RUN_DEFAULT);
499
480 - info("Shutting down ACLK_DATABASE engine event loop complete.");
500 + info("Shutting down ACLK sync event loop complete.");
501 /* TODO: don't let the API block by waiting to enqueue commands */
502 uv_cond_destroy(&wc->cmd_cond);
503 /* uv_mutex_destroy(&wc->cmd_mutex); */
@@ -503,8 +523,6 @@ error_after_async_init:
523 fatal_assert(0 == uv_loop_close(loop));
524 error_after_loop_init:
525 freez(loop);
506 -
507 - wc->error = UV_EAGAIN;
526 }
527
528 // -------------------------------------------------------------
database/sqlite/sqlite_aclk.h
+12 -8
@@ -115,10 +115,18 @@ static inline char *get_str_from_uuid(uuid_t *uuid)
115 enum aclk_database_opcode {
116 ACLK_DATABASE_NOOP = 0,
117 ACLK_DATABASE_ADD_ALERT,
118 +
119 +#ifdef ENABLE_NEW_CLOUD_PROTOCOL
120 ACLK_DATABASE_ADD_CHART,
121 ACLK_DATABASE_ADD_DIMENSION,
120 - ACLK_DATABASE_ALARM_HEALTH_LOG,
122 + ACLK_DATABASE_PUSH_CHART,
123 + ACLK_DATABASE_PUSH_CHART_CONFIG,
124 + ACLK_DATABASE_RESET_CHART,
125 ACLK_DATABASE_CHART_ACK,
126 + ACLK_DATABASE_UPD_RETENTION,
127 + ACLK_DATABASE_DIM_DELETION,
128 +#endif
129 + ACLK_DATABASE_ALARM_HEALTH_LOG,
130 ACLK_DATABASE_CLEANUP,
131 ACLK_DATABASE_DELETE_HOST,
132 ACLK_DATABASE_NODE_INFO,
@@ -126,12 +134,7 @@ enum aclk_database_opcode {
134 ACLK_DATABASE_PUSH_ALERT_CONFIG,
135 ACLK_DATABASE_PUSH_ALERT_SNAPSHOT,
136 ACLK_DATABASE_QUEUE_REMOVED_ALERTS,
129 - ACLK_DATABASE_PUSH_CHART,
130 - ACLK_DATABASE_PUSH_CHART_CONFIG,
131 - ACLK_DATABASE_RESET_CHART,
132 - ACLK_DATABASE_SHUTDOWN,
133 - ACLK_DATABASE_TIMER,
134 - ACLK_DATABASE_UPD_RETENTION
137 + ACLK_DATABASE_TIMER
138 };
139
140 struct aclk_chart_payload_t {
@@ -172,6 +175,7 @@ struct aclk_database_worker_config {
175 uint64_t alerts_batch_id; // batch id for alerts to use
176 uint64_t alerts_start_seq_id; // cloud has asked to start streaming from
177 uint64_t alert_sequence_id; // last alert sequence_id
178 + uint32_t chart_payload_count;
179 uint64_t alerts_snapshot_id; //will contain the snapshot_id value if snapshot was requested
180 uint64_t alerts_ack_sequence_id; //last sequence_id ack'ed from cloud via sendsnapshot message
181 uv_loop_t *loop;
@@ -182,7 +186,7 @@ struct aclk_database_worker_config {
186 uv_cond_t cmd_cond;
187 volatile unsigned queue_size;
188 struct aclk_database_cmdqueue cmd_queue;
185 - int error;
189 + uint32_t retry_count;
190 int chart_updates;
191 int alert_updates;
192 time_t batch_created;
database/sqlite/sqlite_aclk_alert.c
+1 -1
@@ -583,7 +583,7 @@ void sql_queue_removed_alerts_to_aclk(RRDHOST *host)
583 {
584 #ifdef ENABLE_NEW_CLOUD_PROTOCOL
585 if (unlikely(!host->dbsync_worker)) {
586 - error("ACLK synchronization thread is not active for host %s", host->hostname);
586 + error("ACLK synchronization thread is not active for host %s when trying to queue removed alerts", host->hostname);
587 return;
588 }
589
database/sqlite/sqlite_aclk_chart.c
+343 -205
@@ -6,12 +6,11 @@
6 #ifdef ENABLE_NEW_CLOUD_PROTOCOL
7 #include "../../aclk/aclk_charts_api.h"
8 #include "../../aclk/aclk.h"
9 -#endif
9
11 -#ifdef ENABLE_NEW_CLOUD_PROTOCOL
10 static inline int sql_queue_chart_payload(struct aclk_database_worker_config *wc,
11 void *data, enum aclk_database_opcode opcode)
12 {
13 + int rc;
14 if (unlikely(!wc))
15 return 1;
16
@@ -19,8 +18,8 @@ static inline int sql_queue_chart_payload(struct aclk_database_worker_config *wc
18 memset(&cmd, 0, sizeof(cmd));
19 cmd.opcode = opcode;
20 cmd.data = data;
22 - aclk_database_enq_cmd(wc, &cmd);
23 - return 0;
21 + rc = aclk_database_enq_cmd_noblock(wc, &cmd);
22 + return rc;
23 }
24
25 static int payload_sent(char *uuid_str, uuid_t *uuid, void *payload, size_t payload_size)
@@ -60,13 +59,13 @@ bind_fail:
59 return send_status;
60 }
61
63 -static int aclk_add_chart_payload(char *uuid_str, uuid_t *uuid, char *claim_id, ACLK_PAYLOAD_TYPE payload_type,
64 - void *payload, size_t payload_size)
62 +static int aclk_add_chart_payload(struct aclk_database_worker_config *wc, uuid_t *uuid, char *claim_id,
63 + ACLK_PAYLOAD_TYPE payload_type, void *payload, size_t payload_size)
64 {
65 static __thread sqlite3_stmt *res_chart = NULL;
66 int rc;
67
69 - rc = payload_sent(uuid_str, uuid, payload, payload_size);
68 + rc = payload_sent(wc->uuid_str, uuid, payload, payload_size);
69 if (rc == 1)
70 return 0;
71
@@ -74,7 +73,7 @@ static int aclk_add_chart_payload(char *uuid_str, uuid_t *uuid, char *claim_id,
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) " \
77 - "VALUES (@unique_id, @uuid, @claim_id, strftime('%%s','now'), @type, @payload);", uuid_str);
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);
@@ -115,20 +114,25 @@ static int aclk_add_chart_payload(char *uuid_str, uuid_t *uuid, char *claim_id,
114 rc = execute_insert(res_chart);
115 if (unlikely(rc != SQLITE_DONE))
116 error_report("Failed store chart payload event, rc = %d", rc);
117 + else {
118 + wc->chart_payload_count++;
119 + time_t now = now_realtime_sec();
120 + if (wc->rotation_after > now && wc->rotation_after < now + ACLK_DATABASE_ROTATION_DELAY)
121 + wc->rotation_after = now + ACLK_DATABASE_ROTATION_DELAY;
122 + }
123
124 bind_fail:
125 if (unlikely(sqlite3_reset(res_chart) != SQLITE_OK))
126 error_report("Failed to reset statement in store chart payload, rc = %d", rc);
127 return (rc != SQLITE_DONE);
128 }
124 -#endif
129 +
130
131 int aclk_add_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
132 {
133 int rc = 0;
134 CHECK_SQLITE_CONNECTION(db_meta);
135
131 -#ifdef ENABLE_NEW_CLOUD_PROTOCOL
136 char *claim_id = is_agent_claimed();
137
138 RRDSET *st = cmd.data;
@@ -158,14 +162,38 @@ int aclk_add_chart_event(struct aclk_database_worker_config *wc, struct aclk_dat
162 size_t size;
163 char *payload = generate_chart_instance_updated(&size, &chart_payload);
164 if (likely(payload))
161 - rc = aclk_add_chart_payload(wc->uuid_str, st->chart_uuid, claim_id, ACLK_PAYLOAD_CHART, (void *) payload, size);
165 + rc = aclk_add_chart_payload(wc, st->chart_uuid, claim_id, ACLK_PAYLOAD_CHART, (void *) payload, size);
166 freez(payload);
167 chart_instance_updated_destroy(&chart_payload);
168 }
165 -#else
166 - UNUSED(wc);
167 - UNUSED(cmd);
168 -#endif
169 + return rc;
170 +}
171 +
172 +static inline int aclk_upd_dimension_event(struct aclk_database_worker_config *wc, char *claim_id, uuid_t *dim_uuid,
173 + const char *dim_id, const char *dim_name, const char *chart_name, time_t first_time, time_t last_time)
174 +{
175 + int rc = 0;
176 + size_t size;
177 +
178 + if (unlikely(!dim_uuid || !dim_id || !dim_name || !chart_name))
179 + return 0;
180 +
181 + struct chart_dimension_updated dim_payload;
182 + memset(&dim_payload, 0, sizeof(dim_payload));
183 +
184 + if (!first_time)
185 + info("DEBUG: Deleting dimension [%s] [%s] [%s] [%s] [%s]", wc->node_id, claim_id, dim_id, dim_name, chart_name);
186 + dim_payload.node_id = wc->node_id;
187 + dim_payload.claim_id = claim_id;
188 + dim_payload.name = dim_name;
189 + dim_payload.id = dim_id;
190 + dim_payload.chart_id = chart_name;
191 + dim_payload.created_at.tv_sec = first_time;
192 + dim_payload.last_timestamp.tv_sec = last_time;
193 + char *payload = generate_chart_dimension_updated(&size, &dim_payload);
194 + if (likely(payload))
195 + rc = aclk_add_chart_payload(wc, dim_uuid, claim_id, ACLK_PAYLOAD_DIMENSION, (void *)payload, size);
196 + freez(payload);
197 return rc;
198 }
199
@@ -174,7 +202,6 @@ int aclk_add_dimension_event(struct aclk_database_worker_config *wc, struct aclk
202 int rc = 0;
203 CHECK_SQLITE_CONNECTION(db_meta);
204
177 -#ifdef ENABLE_NEW_CLOUD_PROTOCOL
205 char *claim_id = is_agent_claimed();
206
207 RRDDIM *rd = cmd.data;
@@ -187,37 +214,25 @@ int aclk_add_dimension_event(struct aclk_database_worker_config *wc, struct aclk
214
215 int live = ((now - last_t) < (RRDSET_MINIMUM_LIVE_COUNT * rd->update_every));
216
190 - struct chart_dimension_updated dim_payload;
191 - size_t size;
192 -
193 - memset(&dim_payload, 0, sizeof(dim_payload));
194 - dim_payload.node_id = wc->node_id;
195 - dim_payload.claim_id = claim_id;
196 - dim_payload.name = rd->name;
197 - dim_payload.id = rd->id;
217 + rc = aclk_upd_dimension_event(
218 + wc,
219 + claim_id,
220 + &rd->state->metric_uuid,
221 + rd->id,
222 + rd->name,
223 + rd->rrdset->id,
224 + first_t,
225 + live ? 0 : last_t);
226
199 - dim_payload.chart_id = rd->rrdset->name;
200 - dim_payload.created_at.tv_sec = first_t;
201 - if (unlikely(!live))
202 - dim_payload.last_timestamp.tv_sec = last_t;
203 -
204 - char *payload = generate_chart_dimension_updated(&size, &dim_payload);
205 - if (likely(payload))
206 - rc = aclk_add_chart_payload(wc->uuid_str, &rd->state->metric_uuid, claim_id, ACLK_PAYLOAD_DIMENSION, (void *)payload, size);
207 - freez(payload);
227 freez(claim_id);
228 }
229 rrddim_flag_clear(rd, RRDDIM_FLAG_ACLK);
211 -#else
212 - UNUSED(wc);
213 - UNUSED(cmd);
214 -#endif
230 return rc;
231 }
232
233 +
234 void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
235 {
220 -#ifdef ENABLE_NEW_CLOUD_PROTOCOL
236 int rc;
237
238 wc->chart_pending = 0;
@@ -269,6 +284,8 @@ void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_d
284
285 int loop = cmd.param1;
286
287 + uint64_t start_sequence_id = wc->chart_sequence_id;
288 +
289 while (loop > 0) {
290 uint64_t previous_sequence_id = wc->chart_sequence_id;
291 int count = 0;
@@ -294,6 +311,8 @@ void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_d
311 previous_sequence_id = last_sequence;
312 is_dim[count] = sqlite3_column_int(res, 3) > 0;
313 count++;
314 + if (wc->chart_payload_count)
315 + wc->chart_payload_count--;
316 }
317 freez(payload_list[count]);
318 payload_list_max_size[count] = 0;
@@ -325,9 +344,22 @@ void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_d
344 wc->chart_sequence_id = last_sequence;
345 wc->chart_timestamp = last_timestamp;
346 }
347 + else
348 + break;
349 --loop;
350 }
351
352 + if (start_sequence_id != wc->chart_sequence_id) {
353 + time_t now = now_realtime_sec();
354 + if (wc->rotation_after > now && wc->rotation_after < now + ACLK_DATABASE_ROTATION_DELAY)
355 + wc->rotation_after = now + ACLK_DATABASE_ROTATION_DELAY;
356 + }
357 + else {
358 + wc->chart_payload_count = sql_get_pending_count(wc);
359 + if (!wc->chart_payload_count)
360 + info("%s: sync of charts and dimensions done in %ld seconds", wc->host->hostname, now_realtime_sec() - wc->startup_time);
361 + }
362 +
363 for (int i = 0; i <= limit; ++i)
364 freez(payload_list[i]);
365
@@ -344,19 +376,13 @@ bind_fail:
376
377 buffer_free(sql);
378 freez(claim_id);
347 -#else
348 - UNUSED(wc);
349 - UNUSED(cmd);
350 -#endif
379 return;
380 }
381
354 -
382 // Push one chart config to the cloud
383 int aclk_send_chart_config(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
384 {
385 UNUSED(wc);
359 -#ifdef ENABLE_NEW_CLOUD_PROTOCOL
386
387 CHECK_SQLITE_CONNECTION(db_meta);
388
@@ -419,12 +445,9 @@ int aclk_send_chart_config(struct aclk_database_worker_config *wc, struct aclk_d
445 freez((char *) cmd.data_param);
446 buffer_free(sql);
447 return rc;
422 -#else
423 - UNUSED(cmd);
424 - return 0;
425 -#endif
448 }
449
450 +
451 void aclk_receive_chart_ack(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
452 {
453 int rc;
@@ -460,7 +483,6 @@ void aclk_receive_chart_ack(struct aclk_database_worker_config *wc, struct aclk_
483
484 void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
485 {
463 -#ifdef ENABLE_NEW_CLOUD_PROTOCOL
486 BUFFER *sql = buffer_create(1024);
487 buffer_sprintf(sql, "UPDATE aclk_chart_%s SET status = NULL, date_submitted = NULL WHERE sequence_id >= %"PRIu64";",
488 wc->uuid_str, cmd.param1);
@@ -479,35 +501,33 @@ void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct acl
501 db_unlock();
502 wc->chart_sequence_id = 0;
503 wc->chart_timestamp = 0;
504 + wc->chart_payload_count = 0;
505
506 RRDHOST *host = wc->host;
507 rrdhost_rdlock(host);
508 RRDSET *st;
509 rrdset_foreach_read(st, host) {
510 rrdset_rdlock(st);
511 + rrdset_flag_clear(st, RRDSET_FLAG_ACLK);
512 RRDDIM *rd;
513 rrddim_foreach_read(rd, st) {
514 rd->state->aclk_live_status = (rd->state->aclk_live_status == 0);
515 }
492 - rrdset_flag_clear(st, RRDSET_FLAG_ACLK);
516 rrdset_unlock(st);
517 }
518 rrdhost_unlock(host);
519 }
520 else {
498 - //sql_chart_deduplicate(wc, cmd);
499 - sql_get_last_chart_sequence(wc, cmd);
521 + info("Restarting chart sync for %s from sequence=%"PRIu64, wc->uuid_str, cmd.param1);
522 + wc->chart_payload_count = sql_get_pending_count(wc);
523 + sql_get_last_chart_sequence(wc);
524 }
525 buffer_free(sql);
526 wc->chart_updates = 1;
503 -#else
504 - UNUSED(wc);
505 - UNUSED(cmd);
506 -#endif
507 -
527 return;
528 }
529
530 +
531 //
532 // Functions called directly from ACLK threads and will queue commands
533 //
@@ -568,118 +588,282 @@ void aclk_ack_chart_sequence_id(char *node_id, uint64_t last_sequence_id)
588 return;
589 }
590
571 -#ifdef ENABLE_NEW_CLOUD_PROTOCOL
572 -void aclk_reset_chart_event(char *node_id, uint64_t last_sequence_id)
591 +// Start streaming charts / dimensions for node_id
592 +void aclk_start_streaming(char *node_id, uint64_t sequence_id, time_t created_at, uint64_t batch_id)
593 {
594 + UNUSED(created_at);
595 if (unlikely(!node_id))
596 return;
597
577 - debug(D_ACLK_SYNC, "NODE %s wants to resync from %"PRIu64, node_id, last_sequence_id);
578 - aclk_submit_param_command(node_id, ACLK_DATABASE_RESET_CHART, last_sequence_id);
598 + debug(D_ACLK_SYNC,"START streaming charts for node %s from sequence %"PRIu64" t=%ld, batch=%"PRIu64, node_id,
599 + sequence_id, created_at, batch_id);
600 + uuid_t node_uuid;
601 + if (uuid_parse(node_id, node_uuid))
602 + return;
603 +
604 + struct aclk_database_worker_config *wc = NULL;
605 + rrd_wrlock();
606 + RRDHOST *host = localhost;
607 + while(host) {
608 + if (host->node_id && !(uuid_compare(*host->node_id, node_uuid))) {
609 + rrd_unlock();
610 + wc = (struct aclk_database_worker_config *)host->dbsync_worker;
611 + if (likely(wc)) {
612 + wc->chart_reset_count++;
613 + __sync_synchronize();
614 + wc->chart_updates = 0;
615 + wc->batch_id = batch_id;
616 + __sync_synchronize();
617 + wc->batch_created = now_realtime_sec();
618 + if (sequence_id > wc->chart_sequence_id || wc->chart_reset_count > 10) {
619 + info("Requesting full resync from the cloud for node id %s "
620 + "(reset=%d, remote_seq=%"PRIu64", local_seq=%"PRIu64")"
621 + , wc->node_id, wc->chart_reset_count, sequence_id, wc->chart_sequence_id);
622 + chart_reset_t chart_reset;
623 + chart_reset.claim_id = is_agent_claimed();
624 + if (chart_reset.claim_id) {
625 + chart_reset.node_id = node_id;
626 + chart_reset.reason = SEQ_ID_NOT_EXISTS;
627 + aclk_chart_reset(chart_reset);
628 + freez(chart_reset.claim_id);
629 + wc->chart_reset_count = -1;
630 + }
631 + return;
632 + } else {
633 + struct aclk_database_cmd cmd;
634 + memset(&cmd, 0, sizeof(cmd));
635 + // TODO: handle timestamp
636 + if (sequence_id < wc->chart_sequence_id || !sequence_id) { // || created_at != wc->chart_timestamp) {
637 + info("RESET streaming charts for %s from sequence %"PRIu64 \
638 + " t=%ld (reset count=%d)", wc->node_id, wc->chart_sequence_id,
639 + wc->chart_timestamp, wc->chart_reset_count);
640 + cmd.opcode = ACLK_DATABASE_RESET_CHART;
641 + cmd.param1 = sequence_id + 1;
642 + cmd.completion = NULL;
643 + aclk_database_enq_cmd(wc, &cmd);
644 + }
645 + else {
646 + info("START streaming charts for %s enabled -- last streamed sequence %"PRIu64 \
647 + " t=%ld (reset count=%d)", wc->node_id, wc->chart_sequence_id,
648 + wc->chart_timestamp, wc->chart_reset_count);
649 + wc->chart_reset_count = 0;
650 + wc->chart_updates = 1;
651 + }
652 + }
653 + }
654 + else
655 + error("ACLK synchronization thread is not active for host %s", host->hostname);
656 + return;
657 + }
658 + host = host->next;
659 + }
660 + rrd_unlock();
661 return;
662 }
581 -#endif
663
583 -// ST is read locked
584 -int sql_queue_chart_to_aclk(RRDSET *st)
664 +#define SQL_SELECT_HOST_MEMORY_MODE "SELECT memory_mode FROM chart WHERE host_id = @host_id LIMIT 1;"
665 +
666 +static RRD_MEMORY_MODE sql_get_host_memory_mode(uuid_t *host_id)
667 {
586 -#ifdef ENABLE_ACLK
587 -#ifdef ENABLE_NEW_CLOUD_PROTOCOL
588 - if (!aclk_use_new_cloud_arch)
589 -#endif
590 - {
591 - rrdset_flag_clear(st, RRDSET_FLAG_ACLK);
592 - aclk_update_chart(st->rrdhost, st->id, 1);
593 - return 0;
668 + int rc;
669 +
670 + RRD_MEMORY_MODE memory_mode = RRD_MEMORY_MODE_RAM;
671 + sqlite3_stmt *res = NULL;
672 +
673 + rc = sqlite3_prepare_v2(db_meta, SQL_SELECT_HOST_MEMORY_MODE, -1, &res, 0);
674 +
675 + if (unlikely(rc != SQLITE_OK)) {
676 + error_report("Failed to prepare statement to read host memory mode");
677 + return memory_mode;
678 }
595 -#ifdef ENABLE_NEW_CLOUD_PROTOCOL
596 - return sql_queue_chart_payload((struct aclk_database_worker_config *) st->rrdhost->dbsync_worker,
597 - st, ACLK_DATABASE_ADD_CHART);
598 -#else
599 - return 0;
600 -#endif
601 -#else
602 - UNUSED(st);
603 - return 0;
604 -#endif
605 -}
679
607 -int sql_queue_dimension_to_aclk(RRDDIM *rd)
608 -{
609 -#ifdef ENABLE_NEW_CLOUD_PROTOCOL
610 - if (!aclk_use_new_cloud_arch)
611 - return 0;
680 + rc = sqlite3_bind_blob(res, 1, host_id, sizeof(*host_id), SQLITE_STATIC);
681 + if (unlikely(rc != SQLITE_OK)) {
682 + error_report("Failed to bind host parameter to fetch host memory mode");
683 + goto failed;
684 + }
685
613 - int rc = sql_queue_chart_payload((struct aclk_database_worker_config *) rd->rrdset->rrdhost->dbsync_worker,
614 - rd, ACLK_DATABASE_ADD_DIMENSION);
615 - if (likely(!rc))
616 - rrddim_flag_set(rd, RRDDIM_FLAG_ACLK);
617 - return rc;
618 -#else
619 - UNUSED(rd);
620 - return 0;
621 -#endif
686 + while (sqlite3_step(res) == SQLITE_ROW) {
687 + memory_mode = (RRD_MEMORY_MODE) sqlite3_column_int(res, 0);
688 + }
689 +
690 +failed:
691 + rc = sqlite3_finalize(res);
692 + if (unlikely(rc != SQLITE_OK))
693 + error_report("Failed to finalize the prepared statement when reading host memory mode");
694 + return memory_mode;
695 }
696
624 -void sql_chart_deduplicate(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
697 +#define SELECT_HOST_DIMENSION_LIST "SELECT d.dim_id, c.update_every, c.type||'.'||c.id, d.id, d.name FROM chart c, dimension d " \
698 + "WHERE d.chart_id = c.chart_id AND c.host_id = @host_id ORDER BY c.update_every ASC;"
699 +
700 +#define SELECT_HOST_CHART_LIST "SELECT distinct h.host_id, c.update_every, c.type||'.'||c.id FROM chart c, host h " \
701 + "WHERE c.host_id = h.host_id AND c.host_id = @host_id ORDER BY c.update_every ASC;"
702 +
703 +void aclk_update_retention(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
704 {
705 UNUSED(cmd);
706 + int rc;
707
628 - BUFFER *sql = buffer_create(1024);
708 + if (!aclk_use_new_cloud_arch || !aclk_connected)
709 + return;
710
630 - db_lock();
631 - buffer_sprintf(sql, "DROP TABLE IF EXISTS t_%s;", wc->uuid_str);
632 - db_execute(buffer_tostring(sql));
711 + char *claim_id = is_agent_claimed();
712 + if (unlikely(!claim_id))
713 + return;
714
634 - buffer_flush(sql);
635 - buffer_sprintf(sql, "CREATE TABLE t_%s AS SELECT * FROM aclk_chart_payload_%s WHERE unique_id IN "
636 - "(SELECT unique_id from aclk_chart_%s WHERE date_submitted IS NULL AND update_count > 0);",
637 - wc->uuid_str, wc->uuid_str, wc->uuid_str);
638 - db_execute(buffer_tostring(sql));
715 + sqlite3_stmt *res = NULL;
716 + RRD_MEMORY_MODE memory_mode;
717
640 - buffer_flush(sql);
641 - buffer_sprintf(sql, "DELETE FROM aclk_chart_payload_%s WHERE unique_id IN (SELECT unique_id FROM t_%s); " ,
642 - wc->uuid_str, wc->uuid_str);
643 - db_execute(buffer_tostring(sql));
718 + uuid_t host_uuid;
719 + rc = uuid_parse(wc->host_guid, host_uuid);
720 + if (unlikely(rc)) {
721 + freez(claim_id);
722 + return;
723 + }
724
645 - buffer_flush(sql);
646 - buffer_sprintf(sql, "DELETE FROM aclk_chart_%s WHERE unique_id IN (SELECT unique_id FROM t_%s);",
647 - wc->uuid_str, wc->uuid_str);
648 - db_execute(buffer_tostring(sql));
725 + if (wc->host)
726 + memory_mode = wc->host->rrd_memory_mode;
727 + else
728 + memory_mode = sql_get_host_memory_mode(&host_uuid);
729
650 - buffer_flush(sql);
651 - buffer_sprintf(sql, "DELETE FROM aclk_chart_latest_%s WHERE unique_id IN (SELECT unique_id FROM t_%s);",
652 - wc->uuid_str, wc->uuid_str);
653 - db_execute(buffer_tostring(sql));
730 + if (memory_mode == RRD_MEMORY_MODE_DBENGINE)
731 + rc = sqlite3_prepare_v2(db_meta, SELECT_HOST_DIMENSION_LIST, -1, &res, 0);
732 + else
733 + rc = sqlite3_prepare_v2(db_meta, SELECT_HOST_CHART_LIST, -1, &res, 0);
734
655 - buffer_flush(sql);
656 - buffer_sprintf(sql, "INSERT INTO aclk_chart_payload_%s SELECT * FROM t_%s ORDER BY DATE_CREATED ASC;",
657 - wc->uuid_str, wc->uuid_str);
658 - db_execute(buffer_tostring(sql));
735 + if (unlikely(rc != SQLITE_OK)) {
736 + error_report("Failed to prepare statement to fetch host dimensions");
737 + freez(claim_id);
738 + return;
739 + }
740
660 - buffer_flush(sql);
661 - buffer_sprintf(sql, "INSERT OR REPLACE INTO aclk_chart_latest_%s (uuid, unique_id, date_submitted) "
662 - "SELECT uuid, unique_id, date_submitted FROM aclk_chart_%s where sequence_id IN "
663 - "(SELECT sequence_id FROM aclk_chart_%s WHERE date_submitted IS NOT NULL "
664 - "GROUP BY uuid HAVING sequence_id = MAX(sequence_id));"
665 - , wc->uuid_str, wc->uuid_str, wc->uuid_str);
666 - db_execute(buffer_tostring(sql));
741 + rc = sqlite3_bind_blob(res, 1, &host_uuid, sizeof(host_uuid), SQLITE_STATIC);
742 + if (unlikely(rc != SQLITE_OK)) {
743 + error_report("Failed to bind host parameter to fetch host dimensions");
744 + goto failed;
745 + }
746
668 - buffer_flush(sql);
669 - buffer_sprintf(sql, "DROP TABLE IF EXISTS t_%s;", wc->uuid_str);
670 - db_execute(buffer_tostring(sql));
671 - db_unlock();
747 + time_t start_time = LONG_MAX;
748 + time_t first_entry_t;
749 + time_t last_entry_t;
750 + uint32_t update_every = 0;
751
673 - sql_get_last_chart_sequence(wc, cmd);
752 + struct retention_updated rotate_data;
753
675 - buffer_free(sql);
754 + memset(&rotate_data, 0, sizeof(rotate_data));
755 +
756 + int max_intervals = 32;
757 +
758 + rotate_data.interval_duration_count = 0;
759 + rotate_data.interval_durations = callocz(max_intervals, sizeof(*rotate_data.interval_durations));
760 +
761 + now_realtime_timeval(&rotate_data.rotation_timestamp);
762 + rotate_data.memory_mode = memory_mode;
763 + rotate_data.claim_id = claim_id;
764 + rotate_data.node_id = strdupz(wc->node_id);
765 +
766 + // time_t now = now_realtime_sec();
767 + while (sqlite3_step(res) == SQLITE_ROW) {
768 + if (!update_every || update_every != (uint32_t) sqlite3_column_int(res, 1)) {
769 + if (update_every) {
770 + debug(D_ACLK_SYNC,"Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time);
771 + rotate_data.interval_durations[rotate_data.interval_duration_count].retention = rotate_data.rotation_timestamp.tv_sec - start_time;
772 + rotate_data.interval_duration_count++;
773 + }
774 + update_every = (uint32_t) sqlite3_column_int(res, 1);
775 + rotate_data.interval_durations[rotate_data.interval_duration_count].update_every = update_every;
776 + start_time = LONG_MAX;
777 + }
778 +#ifdef ENABLE_DBENGINE
779 + if (memory_mode == RRD_MEMORY_MODE_DBENGINE)
780 + rc = rrdeng_metric_latest_time_by_uuid((uuid_t *)sqlite3_column_blob(res, 0), &first_entry_t, &last_entry_t);
781 + else
782 +#endif
783 + {
784 + if (wc->host) {
785 + RRDSET *st = NULL;
786 + rc = (st = rrdset_find(wc->host, (const char *)sqlite3_column_text(res, 2))) ? 0 : 1;
787 + if (!rc) {
788 + first_entry_t = rrdset_first_entry_t(st);
789 + last_entry_t = rrdset_last_entry_t(st);
790 + }
791 + }
792 + else {
793 + rc = 0;
794 + first_entry_t = rotate_data.rotation_timestamp.tv_sec;
795 + }
796 + }
797 +
798 + if (likely(!rc && first_entry_t))
799 + start_time = MIN(start_time, first_entry_t);
800 +
801 + // if (memory_mode == RRD_MEMORY_MODE_DBENGINE) {
802 + // int live = ((now - last_entry_t) < (RRDSET_MINIMUM_LIVE_COUNT * update_every));
803 + // (void) aclk_upd_dimension_event(
804 + // wc->uuid_str,
805 + // claim_id,
806 + // (uuid_t *)sqlite3_column_blob(res, 0),
807 + // (const char *)(const char *)sqlite3_column_text(res, 3),
808 + // (const char *)(const char *)sqlite3_column_text(res, 4),
809 + // (const char *)(const char *)sqlite3_column_text(res, 2),
810 + // first_entry_t,
811 + // live ? 0 : last_entry_t);
812 + // }
813 + }
814 + if (update_every) {
815 + debug(D_ACLK_SYNC, "Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time);
816 + rotate_data.interval_durations[rotate_data.interval_duration_count].retention = rotate_data.rotation_timestamp.tv_sec - start_time;
817 + rotate_data.interval_duration_count++;
818 + }
819 +
820 + for (int i = 0; i < rotate_data.interval_duration_count; ++i) {
821 +#ifdef NETDATA_INTERNAL_CHECKS
822 + info("%d --> Update %s for %u Retention = %u", i, wc->host_guid,
823 + rotate_data.interval_durations[i].update_every, rotate_data.interval_durations[i].retention);
824 +#endif
825 + };
826 + aclk_retention_updated(&rotate_data);
827 + freez(rotate_data.node_id);
828 + freez(rotate_data.interval_durations);
829 +
830 +failed:
831 + freez(claim_id);
832 + rc = sqlite3_finalize(res);
833 + if (unlikely(rc != SQLITE_OK))
834 + error_report("Failed to finalize the prepared statement when reading host dimensions");
835 return;
836 }
837
679 -void sql_get_last_chart_sequence(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
838 +
839 +uint32_t sql_get_pending_count(struct aclk_database_worker_config *wc)
840 {
681 - UNUSED(cmd);
841 + BUFFER *sql = buffer_create(1024);
842 + sqlite3_stmt *res = NULL;
843 +
844 + buffer_sprintf(sql,"SELECT count(1) FROM aclk_chart_%s ac WHERE ac.date_submitted IS NULL;", wc->uuid_str);
845 +
846 + int rc;
847 + uint32_t chart_payload_count = 0;
848 + rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
849 + if (rc != SQLITE_OK) {
850 + error_report("Failed to prepare statement to count pending messages");
851 + goto fail;
852 + }
853 + while (sqlite3_step(res) == SQLITE_ROW)
854 + chart_payload_count = (uint32_t) sqlite3_column_int(res, 0);
855
856 + rc = sqlite3_finalize(res);
857 + if (unlikely(rc != SQLITE_OK))
858 + error_report("Failed to reset statement when fetching pending messages, rc = %d", rc);
859 +
860 +fail:
861 + buffer_free(sql);
862 + return chart_payload_count;
863 +}
864 +
865 +void sql_get_last_chart_sequence(struct aclk_database_worker_config *wc)
866 +{
867 BUFFER *sql = buffer_create(1024);
868
869 buffer_sprintf(sql,"SELECT ac.sequence_id, ac.date_created FROM aclk_chart_%s ac " \
@@ -711,76 +895,30 @@ fail:
895 return;
896 }
897
714 -// Start streaming charts / dimensions for node_id
715 -void aclk_start_streaming(char *node_id, uint64_t sequence_id, time_t created_at, uint64_t batch_id)
898 +int queue_dimension_to_aclk(RRDDIM *rd)
899 {
717 - UNUSED(created_at);
718 -#ifdef ENABLE_NEW_CLOUD_PROTOCOL
719 - if (unlikely(!node_id))
720 - return;
900 + int rc = sql_queue_chart_payload((struct aclk_database_worker_config *) rd->rrdset->rrdhost->dbsync_worker,
901 + rd, ACLK_DATABASE_ADD_DIMENSION);
902 + return rc;
903 +}
904
722 - debug(D_ACLK_SYNC,"START streaming charts for node %s from sequence %"PRIu64" t=%ld, batch=%"PRIu64, node_id,
723 - sequence_id, created_at, batch_id);
724 - uuid_t node_uuid;
725 - if (uuid_parse(node_id, node_uuid))
726 - return;
905 +#endif //ENABLE_NEW_CLOUD_PROTOCOL
906
728 - struct aclk_database_worker_config *wc = NULL;
729 - rrd_wrlock();
730 - RRDHOST *host = localhost;
731 - while(host) {
732 - if (host->node_id && !(uuid_compare(*host->node_id, node_uuid))) {
733 - rrd_unlock();
734 - wc = (struct aclk_database_worker_config *)host->dbsync_worker;
735 - if (likely(wc)) {
736 - wc->chart_reset_count++;
737 - __sync_synchronize();
738 - wc->chart_updates = 0;
739 - wc->batch_id = batch_id;
740 - __sync_synchronize();
741 - wc->batch_created = now_realtime_sec();
742 - if (sequence_id > wc->chart_sequence_id || wc->chart_reset_count > 10) {
743 - debug(D_ACLK_SYNC,"Requesting full resync from the cloud -- reset_count=%d", wc->chart_reset_count);
744 - chart_reset_t chart_reset;
745 - chart_reset.claim_id = is_agent_claimed();
746 - if (chart_reset.claim_id) {
747 - chart_reset.node_id = node_id;
748 - chart_reset.reason = SEQ_ID_NOT_EXISTS;
749 - aclk_chart_reset(chart_reset);
750 - freez(chart_reset.claim_id);
751 - wc->chart_reset_count = -1;
752 - }
753 - return;
754 - } else {
755 - struct aclk_database_cmd cmd;
756 - memset(&cmd, 0, sizeof(cmd));
757 - // TODO: handle timestamp
758 - if (sequence_id < wc->chart_sequence_id) { // || created_at != wc->chart_timestamp) {
759 - cmd.opcode = ACLK_DATABASE_RESET_CHART;
760 - cmd.param1 = sequence_id + 1;
761 - cmd.completion = NULL;
762 - aclk_database_enq_cmd(wc, &cmd);
763 - }
764 - else {
765 - debug(D_ACLK_SYNC,"START streaming charts for %s enabled -- last streamed sequence %"PRIu64 \
766 - " t=%ld (reset count=%d)", wc->host_guid, wc->chart_sequence_id,
767 - wc->chart_timestamp, wc->chart_reset_count);
768 - wc->chart_reset_count = 0;
769 - wc->chart_updates = 1;
770 - }
771 - }
772 - }
773 - else
774 - error("ACLK synchronization thread is not active for host %s", host->hostname);
775 - return;
776 - }
777 - host = host->next;
778 - }
779 - rrd_unlock();
907 +// ST is read locked
908 +int queue_chart_to_aclk(RRDSET *st)
909 +{
910 +#ifndef ENABLE_NEW_CLOUD_PROTOCOL
911 +#ifdef ENABLE_ACLK
912 + aclk_update_chart(st->rrdhost, st->id, 1);
913 +#endif
914 + return 0;
915 #else
781 - UNUSED(node_id);
782 - UNUSED(sequence_id);
783 - UNUSED(batch_id);
916 + if (!aclk_use_new_cloud_arch && aclk_connected) {
917 + aclk_update_chart(st->rrdhost, st->id, 1);
918 + return 0;
919 + }
920 + return sql_queue_chart_payload((struct aclk_database_worker_config *) st->rrdhost->dbsync_worker,
921 + st, ACLK_DATABASE_ADD_CHART);
922 #endif
785 - return;
923 }
924 +
database/sqlite/sqlite_aclk_chart.h
+4 -8
@@ -16,25 +16,21 @@ extern sqlite3 *db_meta;
16 #define RRDSET_MINIMUM_LIVE_COUNT 3
17 #endif
18
19 -//void aclk_status_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
20 -extern int sql_queue_chart_to_aclk(RRDSET *st);
21 -extern int sql_queue_dimension_to_aclk(RRDDIM *rd);
19 +extern int queue_chart_to_aclk(RRDSET *st);
20 +extern int queue_dimension_to_aclk(RRDDIM *rd);
21 extern void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id);
22 extern void sql_queue_alarm_to_aclk(RRDHOST *host, ALARM_ENTRY *ae);
23 int aclk_add_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
24 int aclk_add_dimension_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
26 -int aclk_add_offline_dimension_event(struct aclk_database_worker_config *wc, char *node_id, char *chart_name, uuid_t *dim_uuid, char *rd_id, char *rd_name, time_t first_entry_t, time_t last_entry_t);
25 int aclk_send_chart_config(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
26 void aclk_ack_chart_sequence_id(char *node_id, uint64_t last_sequence_id);
27 void aclk_get_chart_config(char **hash_id_list);
28 void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
31 -void aclk_reset_chart_event(char *node_id, uint64_t last_sequence_id);
29 void aclk_start_streaming(char *node_id, uint64_t seq_id, time_t created_at, uint64_t batch_id);
30 void sql_chart_deduplicate(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
34 -void sql_check_dimension_state(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
31 void sql_check_rotation_state(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
36 -void sql_get_last_chart_sequence(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
32 +void sql_get_last_chart_sequence(struct aclk_database_worker_config *wc);
33 void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
34 void aclk_receive_chart_ack(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
39 -void sql_update_metric_statistics(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
35 +uint32_t sql_get_pending_count(struct aclk_database_worker_config *wc);
36 #endif //NETDATA_SQLITE_ACLK_CHART_H
database/sqlite/sqlite_aclk_node.c
-159
@@ -5,7 +5,6 @@
5
6 #ifdef ENABLE_NEW_CLOUD_PROTOCOL
7 #include "../../aclk/aclk_charts_api.h"
8 -#include "../../aclk/aclk.h"
8 #endif
9
10 void sql_build_node_info(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
@@ -63,161 +62,3 @@ void sql_build_node_info(struct aclk_database_worker_config *wc, struct aclk_dat
62
63 return;
64 }
66 -#define SQL_SELECT_HOST_MEMORY_MODE "select memory_mode from chart where host_id = @host_id limit 1;"
67 -static RRD_MEMORY_MODE sql_get_host_memory_mode(uuid_t *host_id)
68 -{
69 - int rc;
70 -
71 - RRD_MEMORY_MODE memory_mode = RRD_MEMORY_MODE_RAM;
72 - sqlite3_stmt *res = NULL;
73 -
74 - rc = sqlite3_prepare_v2(db_meta, SQL_SELECT_HOST_MEMORY_MODE, -1, &res, 0);
75 -
76 - if (unlikely(rc != SQLITE_OK)) {
77 - error_report("Failed to prepare statement to read host memory mode");
78 - return memory_mode;
79 - }
80 -
81 - rc = sqlite3_bind_blob(res, 1, host_id, sizeof(*host_id), SQLITE_STATIC);
82 - if (unlikely(rc != SQLITE_OK)) {
83 - error_report("Failed to bind host parameter to fetch host memory mode");
84 - goto failed;
85 - }
86 -
87 - while (sqlite3_step(res) == SQLITE_ROW) {
88 - memory_mode = (RRD_MEMORY_MODE) sqlite3_column_int(res, 0);
89 - }
90 -
91 -failed:
92 - rc = sqlite3_finalize(res);
93 - if (unlikely(rc != SQLITE_OK))
94 - error_report("Failed to finalize the prepared statement when reading host memory mode");
95 - return memory_mode;
96 -}
97 -
98 -#define SELECT_HOST_DIMENSION_LIST "SELECT d.dim_id, c.update_every, c.type||'.'||c.id FROM chart c, dimension d, host h " \
99 - "WHERE d.chart_id = c.chart_id AND c.host_id = h.host_id AND c.host_id = @host_id ORDER BY c.update_every ASC;"
100 -
101 -#define SELECT_HOST_CHART_LIST "SELECT distinct h.host_id, c.update_every, c.type||'.'||c.id FROM chart c, host h " \
102 - "WHERE c.host_id = h.host_id AND c.host_id = @host_id ORDER BY c.update_every ASC;"
103 -
104 -void aclk_update_retention(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
105 -{
106 - UNUSED(cmd);
107 -#ifdef ENABLE_NEW_CLOUD_PROTOCOL
108 - int rc;
109 -
110 - if (!aclk_use_new_cloud_arch || !aclk_connected)
111 - return;
112 -
113 - char *claim_id = is_agent_claimed();
114 - if (unlikely(!claim_id))
115 - return;
116 -
117 - sqlite3_stmt *res = NULL;
118 - RRD_MEMORY_MODE memory_mode;
119 -
120 - uuid_t host_uuid;
121 - rc = uuid_parse(wc->host_guid, host_uuid);
122 - if (unlikely(rc)) {
123 - freez(claim_id);
124 - return;
125 - }
126 -
127 - if (wc->host)
128 - memory_mode = wc->host->rrd_memory_mode;
129 - else
130 - memory_mode = sql_get_host_memory_mode(&host_uuid);
131 -
132 - if (memory_mode == RRD_MEMORY_MODE_DBENGINE)
133 - rc = sqlite3_prepare_v2(db_meta, SELECT_HOST_DIMENSION_LIST, -1, &res, 0);
134 - else
135 - rc = sqlite3_prepare_v2(db_meta, SELECT_HOST_CHART_LIST, -1, &res, 0);
136 -
137 - if (unlikely(rc != SQLITE_OK)) {
138 - error_report("Failed to prepare statement to fetch host dimensions");
139 - freez(claim_id);
140 - return;
141 - }
142 -
143 - rc = sqlite3_bind_blob(res, 1, &host_uuid, sizeof(host_uuid), SQLITE_STATIC);
144 - if (unlikely(rc != SQLITE_OK)) {
145 - error_report("Failed to bind host parameter to fetch host dimensions");
146 - goto failed;
147 - }
148 -
149 - time_t start_time = LONG_MAX;
150 - time_t first_entry_t;
151 - uint32_t update_every = 0;
152 -
153 - struct retention_updated rotate_data;
154 -
155 - memset(&rotate_data, 0, sizeof(rotate_data));
156 -
157 - int max_intervals = 32;
158 -
159 - rotate_data.interval_duration_count = 0;
160 - rotate_data.interval_durations = callocz(max_intervals, sizeof(*rotate_data.interval_durations));
161 -
162 - now_realtime_timeval(&rotate_data.rotation_timestamp);
163 - rotate_data.memory_mode = memory_mode;
164 - rotate_data.claim_id = claim_id;
165 - rotate_data.node_id = strdupz(wc->node_id);
166 -
167 - while (sqlite3_step(res) == SQLITE_ROW) {
168 - if (!update_every || update_every != (uint32_t) sqlite3_column_int(res, 1)) {
169 - if (update_every) {
170 - debug(D_ACLK_SYNC,"Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time);
171 - rotate_data.interval_durations[rotate_data.interval_duration_count].retention = rotate_data.rotation_timestamp.tv_sec - start_time;
172 - rotate_data.interval_duration_count++;
173 - }
174 - update_every = (uint32_t) sqlite3_column_int(res, 1);
175 - rotate_data.interval_durations[rotate_data.interval_duration_count].update_every = update_every;
176 - start_time = LONG_MAX;
177 - }
178 -#ifdef ENABLE_DBENGINE
179 - time_t last_entry_t;
180 - if (memory_mode == RRD_MEMORY_MODE_DBENGINE)
181 - rc = rrdeng_metric_latest_time_by_uuid((uuid_t *)sqlite3_column_blob(res, 0), &first_entry_t, &last_entry_t);
182 - else
183 -#endif
184 - {
185 - if (wc->host) {
186 - RRDSET *st = NULL;
187 - rc = (st = rrdset_find(wc->host, (const char *)sqlite3_column_text(res, 2))) ? 0 : 1;
188 - if (!rc)
189 - first_entry_t = rrdset_first_entry_t(st);
190 - }
191 - else {
192 - rc = 0;
193 - first_entry_t = rotate_data.rotation_timestamp.tv_sec;
194 - }
195 - }
196 -
197 - if (likely(!rc && first_entry_t))
198 - start_time = MIN(start_time, first_entry_t);
199 - }
200 - if (update_every) {
201 - debug(D_ACLK_SYNC, "Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time);
202 - rotate_data.interval_durations[rotate_data.interval_duration_count].retention = rotate_data.rotation_timestamp.tv_sec - start_time;
203 - rotate_data.interval_duration_count++;
204 - }
205 -
206 - for (int i = 0; i < rotate_data.interval_duration_count; ++i) {
207 - debug(D_ACLK_SYNC,"%d --> Update %s for %u Retention = %u", i, wc->host_guid,
208 - rotate_data.interval_durations[i].update_every, rotate_data.interval_durations[i].retention);
209 - };
210 - aclk_retention_updated(&rotate_data);
211 - freez(rotate_data.node_id);
212 - freez(rotate_data.interval_durations);
213 -
214 -failed:
215 - freez(claim_id);
216 - rc = sqlite3_finalize(res);
217 - if (unlikely(rc != SQLITE_OK))
218 - error_report("Failed to finalize the prepared statement when reading host dimensions");
219 -#else
220 - UNUSED(wc);
221 -#endif
222 - return;
223 -}
\ No newline at end of file