Adjust cloud dimension update frequency (#12284)
* Queue a chart immediately to the cloud * Do not inform the cloud immediately if a dimension stopped collecting use MAX(obsoletion time, 1.5 * update_every) * Notify cloud immediately on dimension deletion * Add debug messages * Do not schedule an update if we are shutting down
Stelios Fragkakis committed
Mar 8, 2022 at 20:06 UTC
6872df9e6ac91c4fc8eb56c8736b7ab567c0638a
6 files changed
+62
-10
database/rrd.h
-1
@@ -53,7 +53,6 @@ struct context_param {
53
uint8_t flags;
54
};
55
56
-#define RRDSET_MINIMUM_LIVE_COUNT 3
56
#define META_CHART_UPDATED 1
57
#define META_PLUGIN_UPDATED 2
58
#define META_MODULE_UPDATED 4
database/rrddim.c
+4
@@ -504,6 +504,10 @@ void rrddim_free_custom(RRDSET *st, RRDDIM *rd, int db_rotated)
504
error("RRDDIM: INTERNAL ERROR: attempt to remove from index dimension '%s' on chart '%s', removed a different dimension.", rd->id, st->id);
505
506
// free(rd->annotations);
507
+#if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
508
+ if (!netdata_exit)
509
+ aclk_send_dimension_update(rd);
510
+#endif
511
512
RRD_MEMORY_MODE rrd_memory_mode = rd->rrd_memory_mode;
513
switch(rrd_memory_mode) {
database/rrdhost.c
+5
@@ -1516,6 +1516,11 @@ restart_after_removal:
1516
}
1517
continue;
1518
}
1519
+#if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
1520
+ else {
1521
+ aclk_send_dimension_update(rd);
1522
+ }
1523
+#endif
1524
}
1525
last = rd;
1526
rd = rd->next;
database/rrdset.c
+7
-5
@@ -1400,10 +1400,8 @@ void rrdset_done(RRDSET *st) {
1400
#ifdef ENABLE_ACLK
1401
if (likely(!st->state->is_ar_chart)) {
1402
if (unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
1403
- if (st->counter_done >= RRDSET_MINIMUM_LIVE_COUNT && st->dimensions) {
1404
- if (likely(!queue_chart_to_aclk(st)))
1405
- rrdset_flag_set(st, RRDSET_FLAG_ACLK);
1406
- }
1403
+ if (likely(st->dimensions && !queue_chart_to_aclk(st)))
1404
+ rrdset_flag_set(st, RRDSET_FLAG_ACLK);
1405
}
1406
}
1407
#endif
@@ -1832,7 +1830,8 @@ after_second_database_work:
1830
#if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
1831
if (likely(!st->state->is_ar_chart)) {
1832
if (!rrddim_flag_check(rd, RRDDIM_FLAG_HIDDEN)) {
1835
- int live = ((mark - rd->last_collected_time.tv_sec) < (RRDSET_MINIMUM_LIVE_COUNT * rd->update_every));
1833
+ int live = ((mark - rd->last_collected_time.tv_sec) <
1834
+ MAX(RRDSET_MINIMUM_LIVE_MULTIPLIER * rd->update_every, rrdset_free_obsolete_time));
1835
if (unlikely(live != rd->state->aclk_live_status)) {
1836
if (likely(rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
1837
if (likely(!queue_dimension_to_aclk(rd))) {
@@ -1943,6 +1942,9 @@ after_second_database_work:
1942
delete_dimension_uuid(&rd->state->metric_uuid);
1943
} else {
1944
/* Do not delete this dimension */
1945
+#if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
1946
+ aclk_send_dimension_update(rd);
1947
+#endif
1948
last = rd;
1949
rd = rd->next;
1950
continue;
database/sqlite/sqlite_aclk_chart.c
+43
-2
@@ -3,7 +3,7 @@
3
#include "sqlite_functions.h"
4
#include "sqlite_aclk_chart.h"
5
6
-#ifdef ENABLE_NEW_CLOUD_PROTOCOL
6
+#if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
7
#include "../../aclk/aclk_charts_api.h"
8
#include "../../aclk/aclk.h"
9
@@ -277,7 +277,7 @@ int aclk_add_dimension_event(struct aclk_database_worker_config *wc, struct aclk
277
time_t first_t = rd->state->query_ops.oldest_time(rd);
278
time_t last_t = rd->state->query_ops.latest_time(rd);
279
280
- int live = ((now - last_t) < (RRDSET_MINIMUM_LIVE_COUNT * rd->update_every));
280
+ int live = ((now - last_t) < MAX(RRDSET_MINIMUM_LIVE_MULTIPLIER * rd->update_every, rrdset_free_obsolete_time));
281
282
rc = aclk_upd_dimension_event(
283
wc,
@@ -973,6 +973,47 @@ int queue_dimension_to_aclk(RRDDIM *rd)
973
return rc;
974
}
975
976
+void aclk_send_dimension_update(RRDDIM *rd)
977
+{
978
+ if (!aclk_use_new_cloud_arch)
979
+ return;
980
+
981
+ char *claim_id = is_agent_claimed();
982
+ if (unlikely(!claim_id))
983
+ return;
984
+
985
+ time_t first_entry_t = rrddim_first_entry_t(rd);
986
+ time_t last_entry_t = rrddim_last_entry_t(rd);
987
+
988
+ time_t now = now_realtime_sec();
989
+ int live = ((now - rd->last_collected_time.tv_sec) <
990
+ MAX(RRDSET_MINIMUM_LIVE_MULTIPLIER * rd->update_every, rrdset_free_obsolete_time));
991
+
992
+ if (!live || rd->state->aclk_live_status != live || !first_entry_t) {
993
+ (void)aclk_upd_dimension_event(
994
+ rd->rrdset->rrdhost->dbsync_worker,
995
+ claim_id,
996
+ &rd->state->metric_uuid,
997
+ rd->id,
998
+ rd->name,
999
+ rd->rrdset->id,
1000
+ first_entry_t,
1001
+ live ? 0 : last_entry_t);
1002
+
1003
+ if (!first_entry_t)
1004
+ debug(D_ACLK_SYNC, "%s: Update dimension chart=%s dim=%s live=%d (%ld, %ld)",
1005
+ rd->rrdset->rrdhost->hostname, rd->rrdset->name, rd->name, live, first_entry_t, last_entry_t);
1006
+ else
1007
+ debug(D_ACLK_SYNC, "%s: Update dimension chart=%s dim=%s live=%d (%ld, %ld) collected %ld seconds ago",
1008
+ rd->rrdset->rrdhost->hostname, rd->rrdset->name, rd->name, live, first_entry_t,
1009
+ last_entry_t, now - last_entry_t);
1010
+ rd->state->aclk_live_status = live;
1011
+ }
1012
+
1013
+ freez(claim_id);
1014
+ return;
1015
+}
1016
+
1017
#endif //ENABLE_NEW_CLOUD_PROTOCOL
1018
1019
// ST is read locked
database/sqlite/sqlite_aclk_chart.h
+3
-2
@@ -12,8 +12,8 @@ typedef enum payload_type {
12
13
extern sqlite3 *db_meta;
14
15
-#ifndef RRDSET_MINIMUM_LIVE_COUNT
16
-#define RRDSET_MINIMUM_LIVE_COUNT 3
15
+#ifndef RRDSET_MINIMUM_LIVE_MULTIPLIER
16
+#define RRDSET_MINIMUM_LIVE_MULTIPLIER (1.5)
17
#endif
18
19
extern int queue_chart_to_aclk(RRDSET *st);
@@ -34,4 +34,5 @@ void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct acl
34
void aclk_receive_chart_ack(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
35
void aclk_process_dimension_deletion(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
36
uint32_t sql_get_pending_count(struct aclk_database_worker_config *wc);
37
+void aclk_send_dimension_update(RRDDIM *rd);
38
#endif //NETDATA_SQLITE_ACLK_CHART_H