Improve agent to cloud status update process (#16342)
* Send node update info only if the host has finished replication * Log number of hosts replicating / pending to load context * Remove prefix (thread name is enough)
Stelios Fragkakis committed
Nov 7, 2023 at 11:39 UTC
c78566b2d0eb7c48287e2e1c4ad664e8ac9fdce1
2 files changed
+35
-35
database/sqlite/sqlite_aclk_node.c
+34
-35
@@ -7,6 +7,7 @@
7
#include "../../aclk/aclk_capas.h"
8
9
#ifdef ENABLE_ACLK
10
+
11
DICTIONARY *collectors_from_charts(RRDHOST *host, DICTIONARY *dict) {
12
RRDSET *st;
13
char name[500];
@@ -26,17 +27,9 @@ DICTIONARY *collectors_from_charts(RRDHOST *host, DICTIONARY *dict) {
27
return dict;
28
}
29
29
-static void build_node_collectors(char *node_id __maybe_unused)
30
+static void build_node_collectors(RRDHOST *host)
31
{
31
-
32
- RRDHOST *host = find_host_by_node_id(node_id);
33
-
34
- if (unlikely(!host))
35
- return;
36
-
32
struct aclk_sync_host_config *wc = (struct aclk_sync_host_config *) host->aclk_sync_host_config;
38
- if (unlikely(!wc))
39
- return;
33
34
struct update_node_collectors upd_node_collectors;
35
DICTIONARY *dict = dictionary_create(DICT_OPTION_SINGLE_THREADED);
@@ -50,29 +43,15 @@ static void build_node_collectors(char *node_id __maybe_unused)
43
dictionary_destroy(dict);
44
freez(upd_node_collectors.claim_id);
45
53
- netdata_log_access("ACLK RES [%s (%s)]: NODE COLLECTORS SENT", node_id, rrdhost_hostname(host));
54
-
55
- freez(node_id);
46
+ netdata_log_access("ACLK RES [%s (%s)]: NODE COLLECTORS SENT", wc->node_id, rrdhost_hostname(host));
47
}
48
58
-static void build_node_info(char *node_id __maybe_unused)
49
+static void build_node_info(RRDHOST *host)
50
{
51
struct update_node_info node_info;
52
62
- RRDHOST *host = find_host_by_node_id(node_id);
63
-
64
- if (unlikely((!host))) {
65
- freez(node_id);
66
- return;
67
- }
68
-
53
struct aclk_sync_host_config *wc = (struct aclk_sync_host_config *) host->aclk_sync_host_config;
54
71
- if (unlikely(!wc)) {
72
- freez(node_id);
73
- return;
74
- }
75
-
55
rrd_rdlock();
56
node_info.node_id = wc->node_id;
57
node_info.claim_id = get_agent_claimid();
@@ -132,10 +111,21 @@ static void build_node_info(char *node_id __maybe_unused)
111
freez(host_version);
112
113
wc->node_collectors_send = now_realtime_sec();
135
- freez(node_id);
136
-
114
}
115
116
+bool host_is_replicating(RRDHOST *host)
117
+{
118
+ bool replicating = false;
119
+ RRDSET *st;
120
+ rrdset_foreach_reentrant(st, host) {
121
+ if (rrdset_is_replicating(st)) {
122
+ replicating = true;
123
+ break;
124
+ }
125
+ }
126
+ rrdset_foreach_done(st);
127
+ return replicating;
128
+}
129
130
void aclk_check_node_info_and_collectors(void)
131
{
@@ -144,35 +134,44 @@ void aclk_check_node_info_and_collectors(void)
134
if (unlikely(!aclk_connected))
135
return;
136
147
- size_t pending = 0;
148
- dfe_start_reentrant(rrdhost_root_index, host) {
149
-
137
+ size_t context_loading = 0;
138
+ size_t replicating = 0;
139
+ dfe_start_reentrant(rrdhost_root_index, host)
140
+ {
141
struct aclk_sync_host_config *wc = host->aclk_sync_host_config;
142
if (unlikely(!wc))
143
continue;
144
145
if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))) {
146
internal_error(true, "ACLK SYNC: Context still pending for %s", rrdhost_hostname(host));
156
- pending++;
147
+ context_loading++;
148
+ continue;
149
+ }
150
+
151
+ if (unlikely(host_is_replicating(host))) {
152
+ internal_error(true, "ACLK SYNC: Host %s is still replicating", rrdhost_hostname(host));
153
+ replicating++;
154
continue;
155
}
156
157
if (wc->node_info_send_time && wc->node_info_send_time + 30 < now_realtime_sec()) {
158
wc->node_info_send_time = 0;
162
- build_node_info(strdupz(wc->node_id));
159
+ build_node_info(host);
160
internal_error(true, "ACLK SYNC: Sending node info for %s", rrdhost_hostname(host));
161
}
162
163
if (wc->node_collectors_send && wc->node_collectors_send + 30 < now_realtime_sec()) {
167
- build_node_collectors(strdupz(wc->node_id));
164
+ build_node_collectors(host);
165
internal_error(true, "ACLK SYNC: Sending collectors for %s", rrdhost_hostname(host));
166
wc->node_collectors_send = 0;
167
}
168
}
169
dfe_done(host);
170
174
- if(pending)
175
- netdata_log_info("ACLK: %zu nodes are pending for contexts to load, skipped sending node info for them", pending);
171
+ if (context_loading || replicating) {
172
+ error_limit_static_thread_var(erl, 10, 100 * USEC_PER_MS);
173
+ error_limit(&erl, "%zu nodes loading contexts, %zu replicating data", context_loading, replicating);
174
+ }
175
}
176
177
#endif
database/sqlite/sqlite_aclk_node.h
+1
@@ -4,4 +4,5 @@
4
#define NETDATA_SQLITE_ACLK_NODE_H
5
6
void aclk_check_node_info_and_collectors(void);
7
+bool host_finished_replication(RRDHOST *host);
8
#endif //NETDATA_SQLITE_ACLK_NODE_H