@cryptotaxi247 / netdata-1 / commits / 4179fa8ff

Streaming improvements No 6 (#19196)

* pulse aral stats fixes * double is not needed there * control the number of pages evicted at once, to avoid gaps on data collection * lower the aggresiveness of evictions to prevent evicting signifanctly more pages than really needed * add aral judy statistics to pulse aral * metadata sync shutsdown in the background * renumber * when just 1 host is still pending, log also its hostname * call mallocz_release_as_much_memory_to_the_system() at most once per second * improve ML logs and move metadata_sync_shutdown() at the proper place * stack size of external plugins; moved all function evloop data off the stack * fix for incorrect replication accounting * metadata cmds in aral * fix function payload parsing for external plugins * split main service control and netdata shutdown to separate files - no code changes * add more information to shutdown watcher * fix exit log messages * parallel dbengine exits for all tiers * log with limit * alerts not alarms * minor fixes in health * added logs to trace the root cause of delays when closing data files * added STORAGE_PRIORITY_SYNCHRONOUS_FIRST * Revert "added logs to trace the root cause of delays when closing data files" This reverts commit 75515c148421102ad1942dc340fe0391a4e012cc. * log datafile lock wait * print the number of writers * print the number of writers again * print the number of writers again again * fix watcher messages * single node agents use 1 replication and 1 ML thread; max threads for both is 256 * log the progress of flushers * spawn intense flushing on quiesce * cleanup logs added * print the status of the cache on exit * dbengine shutdown cleanup * cleanup logs * cleanup logs * more cleanup on logs * proper percentage calculation * sentry include added

Costa Tsaousis committed Dec 14, 2024 at 00:35 UTC 4179fa8ffd14fdeb1ea6d7bd7bae88a1792f60b7

69 files changed +1016 -904

CMakeLists.txt

+6 -2

@@ -1051,8 +1051,8 @@ set(DAEMON_FILES

1051 src/daemon/signals.c

1052 src/daemon/signals.h

1053 src/daemon/service.c

1054 - src/daemon/watcher.c

1055 - src/daemon/watcher.h

1054 + src/daemon/daemon-shutdown-watcher.c

1055 + src/daemon/daemon-shutdown-watcher.h

1056 src/daemon/static_threads.c

1057 src/daemon/static_threads.h

1058 src/daemon/commands.c

@@ -1114,6 +1114,10 @@ set(DAEMON_FILES

1114 src/daemon/config/netdata-conf-global.c

1115 src/daemon/config/netdata-conf-global.h

1116 src/daemon/config/netdata-conf.c

1117 + src/daemon/daemon-shutdown.c

1118 + src/daemon/daemon-shutdown.h

1119 + src/daemon/daemon-service.c

1120 + src/daemon/daemon-service.h

1121 )

1122

1123 set(H2O_FILES

src/collectors/apps.plugin/apps_plugin.c

@@ -666,6 +666,7 @@ static bool apps_plugin_exit = false;

666

667 int main(int argc, char **argv) {

668 nd_log_initialize_for_external_plugins("apps.plugin");

669 + netdata_threads_init_for_external_plugins(0);

670

671 pagesize = (size_t)sysconf(_SC_PAGESIZE);

672

src/collectors/cgroups.plugin/sys_fs_cgroup.c

-1

@@ -1327,7 +1327,6 @@ static void cgroup_main_cleanup(void *pptr) {

1327

1328 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

1329

1330 - collector_info("cleaning up...");

1330 worker_unregister();

1331

1332 usec_t max = 2 * USEC_PER_SEC, step = 50000;

src/collectors/cups.plugin/cups_plugin.c

@@ -227,6 +227,7 @@ void reset_metrics() {

227

228 int main(int argc, char **argv) {

229 nd_log_initialize_for_external_plugins("cups.plugin");

230 + netdata_threads_init_for_external_plugins(0);

231

232 parse_command_line(argc, argv);

233

src/collectors/debugfs.plugin/debugfs_plugin.c

@@ -160,6 +160,7 @@ static void debugfs_parse_args(int argc, char **argv)

160 int main(int argc, char **argv)

161 {

162 nd_log_initialize_for_external_plugins("debugfs.plugin");

163 + netdata_threads_init_for_external_plugins(0);

164

165 netdata_configured_host_prefix = getenv("NETDATA_HOST_PREFIX");

166 if (verify_netdata_host_prefix(true) == -1)

src/collectors/diskspace.plugin/plugin_diskspace.c

-4

@@ -516,8 +516,6 @@ static void diskspace_slow_worker_cleanup(void *pptr) {

516 struct slow_worker_data *data = CLEANUP_FUNCTION_GET_PTR(pptr);

517 if(data) return;

518

519 - collector_info("cleaning up...");

520 -

519 worker_unregister();

520 }

521

@@ -608,8 +606,6 @@ static void diskspace_main_cleanup(void *pptr) {

606

607 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

608

611 - collector_info("cleaning up...");

612 -

609 rrd_collector_finished();

610 worker_unregister();

611

src/collectors/ebpf.plugin/ebpf.c

@@ -4006,6 +4006,7 @@ static void ebpf_manage_pid(pid_t pid)

4006 int main(int argc, char **argv)

4007 {

4008 nd_log_initialize_for_external_plugins(NETDATA_EBPF_PLUGIN_NAME);

4009 + netdata_threads_init_for_external_plugins(0);

4010

4011 ebpf_set_global_variables();

4012 if (ebpf_can_plugin_load_code(running_on_kernel, NETDATA_EBPF_PLUGIN_NAME))

src/collectors/freebsd.plugin/plugin_freebsd.c

-1

@@ -78,7 +78,6 @@ static void freebsd_main_cleanup(void *pptr)

79 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

81 - collector_info("cleaning up...");

81 worker_unregister();

83 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

src/collectors/idlejitter.plugin/plugin_idlejitter.c

-1

@@ -10,7 +10,6 @@ static void cpuidlejitter_main_cleanup(void *pptr) {

11 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

13 - collector_info("cleaning up...");

13 worker_unregister();

15 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

src/collectors/macos.plugin/plugin_macos.c

-1

@@ -32,7 +32,6 @@ static void macos_main_cleanup(void *pptr)

33 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

35 - collector_info("cleaning up...");

35 worker_unregister();

37 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

src/collectors/network-viewer.plugin/network-viewer.c

@@ -960,6 +960,7 @@ close_and_send:

960 int main(int argc __maybe_unused, char **argv __maybe_unused) {

961 nd_thread_tag_set("NETWORK-VIEWER");

962 nd_log_initialize_for_external_plugins("network-viewer.plugin");

963 + netdata_threads_init_for_external_plugins(0);

964

965 netdata_configured_host_prefix = getenv("NETDATA_HOST_PREFIX");

966 if(verify_netdata_host_prefix(true) == -1) exit(1);

src/collectors/nfacct.plugin/plugin_nfacct.c

@@ -748,6 +748,7 @@ void nfacct_signals()

748

749 int main(int argc, char **argv) {

750 nd_log_initialize_for_external_plugins("nfacct.plugin");

751 + netdata_threads_init_for_external_plugins(0);

752

753 // ------------------------------------------------------------------------

754 // parse command line parameters

src/collectors/perf.plugin/perf_plugin.c

@@ -1288,6 +1288,7 @@ void parse_command_line(int argc, char **argv) {

1288

1289 int main(int argc, char **argv) {

1290 nd_log_initialize_for_external_plugins("perf.plugin");

1291 + netdata_threads_init_for_external_plugins(0);

1292

1293 parse_command_line(argc, argv);

1294

src/collectors/proc.plugin/plugin_proc.c

-2

@@ -94,8 +94,6 @@ static void proc_main_cleanup(void *pptr)

95 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

97 - collector_info("cleaning up...");

98 -

97 nd_thread_join(netdev_thread);

98 worker_unregister();

src/collectors/proc.plugin/proc_net_dev.c

-2

@@ -1685,8 +1685,6 @@ static void netdev_main_cleanup(void *pptr) {

1685 if(CLEANUP_FUNCTION_GET_PTR(pptr) != (void *)0x01)

1686 return;

1687

1688 - collector_info("cleaning up...");

1689 -

1688 worker_unregister();

1689 }

1690

src/collectors/profile.plugin/plugin_profile.cc

-2

@@ -186,8 +186,6 @@ static void profile_main_cleanup(void *pptr) {

186

187 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

188

189 - netdata_log_info("cleaning up...");

190 -

189 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

190 }

191

src/collectors/slabinfo.plugin/slabinfo.c

@@ -346,6 +346,7 @@ void usage(void) {

346

347 int main(int argc, char **argv) {

348 nd_log_initialize_for_external_plugins("slabinfo.plugin");

349 + netdata_threads_init_for_external_plugins(0);

350

351 program_name = argv[0];

352 int update_every = 1, i, n, freq = 0;

src/collectors/statsd.plugin/statsd.c

-3

@@ -1086,8 +1086,6 @@ void statsd_collector_thread_cleanup(void *pptr) {

1086 d->status->running = false;

1087 spinlock_unlock(&d->status->spinlock);

1088

1089 - collector_info("cleaning up...");

1090 -

1089 #ifdef HAVE_RECVMMSG

1090 size_t i;

1091 for (i = 0; i < d->size; i++)

@@ -2399,7 +2397,6 @@ static void statsd_main_cleanup(void *pptr) {

2397 if(!static_thread) return;

2398

2399 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

2402 - collector_info("cleaning up...");

2400

2401 if (statsd.collection_threads_status) {

2402 int i;

src/collectors/systemd-journal.plugin/systemd-main.c

@@ -20,6 +20,7 @@ static bool journal_data_directories_exist() {

20 int main(int argc __maybe_unused, char **argv __maybe_unused) {

21 nd_thread_tag_set("sd-jrnl.plugin");

22 nd_log_initialize_for_external_plugins("systemd-journal.plugin");

23 + netdata_threads_init_for_external_plugins(0);

25 netdata_configured_host_prefix = getenv("NETDATA_HOST_PREFIX");

26 if(verify_netdata_host_prefix(true) == -1) exit(1);

src/collectors/tc.plugin/plugin_tc.c

-2

@@ -845,8 +845,6 @@ static void tc_main_cleanup(void *pptr) {

845

846 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

847

848 - collector_info("cleaning up...");

849 -

848 if(tc_child_instance) {

849 collector_info("TC: stopping the running tc-qos-helper script");

850 int code = spawn_popen_wait(tc_child_instance); (void)code;

src/collectors/timex.plugin/plugin_timex.c

-1

@@ -37,7 +37,6 @@ static void timex_main_cleanup(void *pptr)

38 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

40 - netdata_log_info("cleaning up...");

40 worker_unregister();

42 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

src/collectors/windows-events.plugin/windows-events.c

@@ -1292,6 +1292,7 @@ void function_windows_events(const char *transaction, char *function, usec_t *st

1292 int main(int argc __maybe_unused, char **argv __maybe_unused) {

1293 nd_thread_tag_set("wevt.plugin");

1294 nd_log_initialize_for_external_plugins("windows-events.plugin");

1295 + netdata_threads_init_for_external_plugins(0);

1296

1297 // ------------------------------------------------------------------------

1298 // initialization

src/collectors/windows.plugin/windows_plugin.c

-2

@@ -48,8 +48,6 @@ static void windows_main_cleanup(void *pptr) {

49 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

51 - collector_info("cleaning up...");

52 -

51 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

53 worker_unregister();

src/collectors/xenstat.plugin/xenstat_plugin.c

@@ -926,6 +926,7 @@ int main(int argc, char **argv) {

926 program_name = PLUGIN_XENSTAT_NAME;

927

928 nd_log_initialize_for_external_plugins(PLUGIN_XENSTAT_NAME);

929 + netdata_threads_init_for_external_plugins(0);

930

931 // ------------------------------------------------------------------------

932 // parse command line parameters

src/daemon/analytics.c

-1

@@ -549,7 +549,6 @@ void analytics_main_cleanup(void *pptr)

549

550 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

551

552 - netdata_log_debug(D_ANALYTICS, "Cleaning up...");

552 analytics_free_data();

553

554 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

src/daemon/config/netdata-conf.c

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

3 #include "netdata-conf.h"

5 +struct config netdata_config = APPCONFIG_INITIALIZER;

6 +

7 bool netdata_conf_load(char *filename, char overwrite_used, const char **user) {

8 static bool run = false;

9 if(run) return false;

src/daemon/config/netdata-conf.h

@@ -5,6 +5,7 @@

6 #include "libnetdata/libnetdata.h"

8 +extern struct config netdata_config;

9 bool netdata_conf_load(char *filename, char overwrite_used, const char **user);

11 #include "netdata-conf-backwards-compatibility.h"

src/daemon/daemon-service.c new

+284

@@ -0,0 +1,284 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#include "daemon-service.h"

4 +

5 +typedef struct service_thread {

6 + pid_t tid;

7 + SERVICE_THREAD_TYPE type;

8 + SERVICE_TYPE services;

9 + char name[ND_THREAD_TAG_MAX + 1];

10 + bool stop_immediately;

11 + bool cancelled;

12 +

13 + union {

14 + ND_THREAD *netdata_thread;

15 + uv_thread_t uv_thread;

16 + };

17 +

18 + force_quit_t force_quit_callback;

19 + request_quit_t request_quit_callback;

20 + void *data;

21 +} SERVICE_THREAD;

22 +

23 +struct service_globals {

24 + SPINLOCK lock;

25 + Pvoid_t pid_judy;

26 +} service_globals = {

27 + .pid_judy = NULL,

28 +};

29 +

30 +SERVICE_THREAD *service_register(SERVICE_THREAD_TYPE thread_type, request_quit_t request_quit_callback, force_quit_t force_quit_callback, void *data, bool update __maybe_unused) {

31 + SERVICE_THREAD *sth = NULL;

32 + pid_t tid = gettid_cached();

33 +

34 + spinlock_lock(&service_globals.lock);

35 + Pvoid_t *PValue = JudyLIns(&service_globals.pid_judy, tid, PJE0);

36 + if(!*PValue) {

37 + sth = callocz(1, sizeof(SERVICE_THREAD));

38 + sth->tid = tid;

39 + sth->type = thread_type;

40 + sth->request_quit_callback = request_quit_callback;

41 + sth->force_quit_callback = force_quit_callback;

42 + sth->data = data;

43 + *PValue = sth;

44 +

45 + switch(thread_type) {

46 + default:

47 + case SERVICE_THREAD_TYPE_NETDATA:

48 + sth->netdata_thread = nd_thread_self();

49 + break;

50 +

51 + case SERVICE_THREAD_TYPE_EVENT_LOOP:

52 + case SERVICE_THREAD_TYPE_LIBUV:

53 + sth->uv_thread = uv_thread_self();

54 + break;

55 + }

56 +

57 + const char *name = nd_thread_tag();

58 + if(!name) name = "";

59 + strncpyz(sth->name, name, sizeof(sth->name) - 1);

60 + }

61 + else {

62 + sth = *PValue;

63 + }

64 + spinlock_unlock(&service_globals.lock);

65 +

66 + return sth;

67 +}

68 +

69 +void service_exits(void) {

70 + pid_t tid = gettid_cached();

71 +

72 + spinlock_lock(&service_globals.lock);

73 + Pvoid_t *PValue = JudyLGet(service_globals.pid_judy, tid, PJE0);

74 + if(PValue) {

75 + freez(*PValue);

76 + JudyLDel(&service_globals.pid_judy, tid, PJE0);

77 + }

78 + spinlock_unlock(&service_globals.lock);

79 +}

80 +

81 +bool service_running(SERVICE_TYPE service) {

82 + static __thread SERVICE_THREAD *sth = NULL;

83 +

84 + if(unlikely(!sth))

85 + sth = service_register(SERVICE_THREAD_TYPE_NETDATA, NULL, NULL, NULL, false);

86 +

87 + sth->services |= service;

88 +

89 + bool cancelled = false;

90 + if (sth->type == SERVICE_THREAD_TYPE_NETDATA)

91 + cancelled = nd_thread_signaled_to_cancel();

92 +

93 + return !sth->stop_immediately && !netdata_exit && !cancelled;

94 +}

95 +

96 +void service_signal_exit(SERVICE_TYPE service) {

97 + spinlock_lock(&service_globals.lock);

98 +

99 + Pvoid_t *PValue;

100 + Word_t tid = 0;

101 + bool first = true;

102 + while((PValue = JudyLFirstThenNext(service_globals.pid_judy, &tid, &first))) {

103 + SERVICE_THREAD *sth = *PValue;

104 +

105 + if((sth->services & service)) {

106 + sth->stop_immediately = true;

107 +

108 + switch(sth->type) {

109 + default:

110 + case SERVICE_THREAD_TYPE_NETDATA:

111 + nd_thread_signal_cancel(sth->netdata_thread);

112 + break;

113 +

114 + case SERVICE_THREAD_TYPE_EVENT_LOOP:

115 + case SERVICE_THREAD_TYPE_LIBUV:

116 + break;

117 + }

118 +

119 + if(sth->request_quit_callback) {

120 + spinlock_unlock(&service_globals.lock);

121 + sth->request_quit_callback(sth->data);

122 + spinlock_lock(&service_globals.lock);

123 + }

124 + }

125 + }

126 +

127 + spinlock_unlock(&service_globals.lock);

128 +}

129 +

130 +static void service_to_buffer(BUFFER *wb, SERVICE_TYPE service) {

131 + if(service & SERVICE_MAINTENANCE)

132 + buffer_strcat(wb, "MAINTENANCE ");

133 + if(service & SERVICE_COLLECTORS)

134 + buffer_strcat(wb, "COLLECTORS ");

135 + if(service & SERVICE_REPLICATION)

136 + buffer_strcat(wb, "REPLICATION ");

137 + if(service & ABILITY_DATA_QUERIES)

138 + buffer_strcat(wb, "DATA_QUERIES ");

139 + if(service & ABILITY_WEB_REQUESTS)

140 + buffer_strcat(wb, "WEB_REQUESTS ");

141 + if(service & SERVICE_WEB_SERVER)

142 + buffer_strcat(wb, "WEB_SERVER ");

143 + if(service & SERVICE_ACLK)

144 + buffer_strcat(wb, "ACLK ");

145 + if(service & SERVICE_HEALTH)

146 + buffer_strcat(wb, "HEALTH ");

147 + if(service & SERVICE_STREAMING)

148 + buffer_strcat(wb, "STREAMING ");

149 + if(service & ABILITY_STREAMING_CONNECTIONS)

150 + buffer_strcat(wb, "STREAMING_CONNECTIONS ");

151 + if(service & SERVICE_CONTEXT)

152 + buffer_strcat(wb, "CONTEXT ");

153 + if(service & SERVICE_ANALYTICS)

154 + buffer_strcat(wb, "ANALYTICS ");

155 + if(service & SERVICE_EXPORTERS)

156 + buffer_strcat(wb, "EXPORTERS ");

157 + if(service & SERVICE_HTTPD)

158 + buffer_strcat(wb, "HTTPD ");

159 +}

160 +

161 +bool service_wait_exit(SERVICE_TYPE service, usec_t timeout_ut) {

162 + BUFFER *service_list = buffer_create(1024, NULL);

163 + BUFFER *thread_list = buffer_create(1024, NULL);

164 + usec_t started_ut = now_monotonic_usec(), ended_ut;

165 + size_t running;

166 + SERVICE_TYPE running_services = 0;

167 +

168 + // cancel the threads

169 + running = 0;

170 + running_services = 0;

171 + {

172 + buffer_flush(thread_list);

173 +

174 + spinlock_lock(&service_globals.lock);

175 +

176 + Pvoid_t *PValue;

177 + Word_t tid = 0;

178 + bool first = true;

179 + while((PValue = JudyLFirstThenNext(service_globals.pid_judy, &tid, &first))) {

180 + SERVICE_THREAD *sth = *PValue;

181 + if(sth->services & service && sth->tid != gettid_cached() && !sth->cancelled) {

182 + sth->cancelled = true;

183 +

184 + switch(sth->type) {

185 + default:

186 + case SERVICE_THREAD_TYPE_NETDATA:

187 + nd_thread_signal_cancel(sth->netdata_thread);

188 + break;

189 +

190 + case SERVICE_THREAD_TYPE_EVENT_LOOP:

191 + case SERVICE_THREAD_TYPE_LIBUV:

192 + break;

193 + }

194 +

195 + if(running)

196 + buffer_strcat(thread_list, ", ");

197 +

198 + buffer_sprintf(thread_list, "'%s' (%d)", sth->name, sth->tid);

199 +

200 + running++;

201 + running_services |= sth->services & service;

202 +

203 + if(sth->force_quit_callback) {

204 + spinlock_unlock(&service_globals.lock);

205 + sth->force_quit_callback(sth->data);

206 + spinlock_lock(&service_globals.lock);

207 + continue;

208 + }

209 + }

210 + }

211 +

212 + spinlock_unlock(&service_globals.lock);

213 + }

214 +

215 + service_signal_exit(service);

216 +

217 + // signal them to stop

218 + size_t last_running = 0;

219 + size_t stale_time_ut = 0;

220 + usec_t sleep_ut = 50 * USEC_PER_MS;

221 + size_t log_countdown_ut = sleep_ut;

222 + do {

223 + if(running != last_running)

224 + stale_time_ut = 0;

225 +

226 + last_running = running;

227 + running = 0;

228 + running_services = 0;

229 + buffer_flush(thread_list);

230 +

231 + spinlock_lock(&service_globals.lock);

232 +

233 + Pvoid_t *PValue;

234 + Word_t tid = 0;

235 + bool first = true;

236 + while((PValue = JudyLFirstThenNext(service_globals.pid_judy, &tid, &first))) {

237 + SERVICE_THREAD *sth = *PValue;

238 + if(sth->services & service && sth->tid != gettid_cached()) {

239 + if(running)

240 + buffer_strcat(thread_list, ", ");

241 +

242 + buffer_sprintf(thread_list, "'%s' (%d)", sth->name, sth->tid);

243 +

244 + running_services |= sth->services & service;

245 + running++;

246 + }

247 + }

248 +

249 + spinlock_unlock(&service_globals.lock);

250 +

251 + if(running) {

252 + log_countdown_ut -= (log_countdown_ut >= sleep_ut) ? sleep_ut : log_countdown_ut;

253 + if(log_countdown_ut == 0 || running != last_running) {

254 + log_countdown_ut = 20 * sleep_ut;

255 +

256 + buffer_flush(service_list);

257 + service_to_buffer(service_list, running_services);

258 + netdata_log_info("SERVICE CONTROL: waiting for the following %zu services [ %s] to exit: %s",

259 + running, buffer_tostring(service_list),

260 + running <= 10 ? buffer_tostring(thread_list) : "");

261 + }

262 +

263 + sleep_usec(sleep_ut);

264 + stale_time_ut += sleep_ut;

265 + }

266 +

267 + ended_ut = now_monotonic_usec();

268 + } while(running && (ended_ut - started_ut < timeout_ut || stale_time_ut < timeout_ut));

269 +

270 + if(running) {

271 + buffer_flush(service_list);

272 + service_to_buffer(service_list, running_services);

273 + netdata_log_info("SERVICE CONTROL: "

274 + "the following %zu service(s) [ %s] take too long to exit: %s; "

275 + "giving up on them...",

276 + running, buffer_tostring(service_list),

277 + buffer_tostring(thread_list));

278 + }

279 +

280 + buffer_free(thread_list);

281 + buffer_free(service_list);

282 +

283 + return (running == 0);

284 +}

src/daemon/daemon-service.h new

+41

@@ -0,0 +1,41 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#ifndef NETDATA_DAEMON_SERVICE_H

4 +#define NETDATA_DAEMON_SERVICE_H

5 +

6 +#include "libnetdata/libnetdata.h"

7 +

8 +typedef enum {

9 + ABILITY_DATA_QUERIES = (1 << 0),

10 + ABILITY_WEB_REQUESTS = (1 << 1),

11 + ABILITY_STREAMING_CONNECTIONS = (1 << 2),

12 + SERVICE_MAINTENANCE = (1 << 3),

13 + SERVICE_COLLECTORS = (1 << 4),

14 + SERVICE_REPLICATION = (1 << 5),

15 + SERVICE_WEB_SERVER = (1 << 6),

16 + SERVICE_ACLK = (1 << 7),

17 + SERVICE_HEALTH = (1 << 8),

18 + SERVICE_STREAMING = (1 << 9),

19 + SERVICE_CONTEXT = (1 << 10),

20 + SERVICE_ANALYTICS = (1 << 11),

21 + SERVICE_EXPORTERS = (1 << 12),

22 + SERVICE_HTTPD = (1 << 13)

23 +} SERVICE_TYPE;

24 +

25 +typedef enum {

26 + SERVICE_THREAD_TYPE_NETDATA,

27 + SERVICE_THREAD_TYPE_LIBUV,

28 + SERVICE_THREAD_TYPE_EVENT_LOOP,

29 +} SERVICE_THREAD_TYPE;

30 +

31 +typedef void (*force_quit_t)(void *data);

32 +typedef void (*request_quit_t)(void *data);

33 +

34 +void service_exits(void);

35 +bool service_running(SERVICE_TYPE service);

36 +struct service_thread *service_register(SERVICE_THREAD_TYPE thread_type, request_quit_t request_quit_callback, force_quit_t force_quit_callback, void *data, bool update __maybe_unused);

37 +

38 +void service_signal_exit(SERVICE_TYPE service);

39 +bool service_wait_exit(SERVICE_TYPE service, usec_t timeout_ut);

40 +

41 +#endif //NETDATA_DAEMON_SERVICE_H

src/daemon/daemon-shutdown-watcher.c renamed

+52 -41

@@ -1,6 +1,6 @@

1 // SPDX-License-Identifier: GPL-3.0-or-later

3 -#include "watcher.h"

3 +#include "daemon-shutdown-watcher.h"

5 watcher_step_t *watcher_steps;

@@ -20,15 +20,28 @@ void watcher_step_complete(watcher_step_id_t step_id) {

20 completion_mark_complete(&watcher_steps[step_id].p);

21 }

23 -static void watcher_wait_for_step(const watcher_step_id_t step_id)

23 +static void watcher_wait_for_step(const watcher_step_id_t step_id, usec_t shutdown_start_time)

24 {

25 - unsigned timeout = 90;

26 -

25 usec_t step_start_time = now_monotonic_usec();

26 + usec_t step_start_duration = step_start_time - shutdown_start_time;

27 +

28 + char start_duration_txt[64];

29 + duration_snprintf(

30 + start_duration_txt, sizeof(start_duration_txt), (int64_t)step_start_duration, "us", true);

31 +

32 + netdata_log_info("shutdown step: [%d/%d] - {at %s} started '%s'...",

33 + (int)step_id + 1, (int)WATCHER_STEP_ID_MAX, start_duration_txt,

34 + watcher_steps[step_id].msg);

36 #ifdef ENABLE_SENTRY

37 // Wait with a timeout

31 - bool ok = completion_timedwait_for(&watcher_steps[step_id].p, timeout);

38 + time_t timeout = 135; // systemd gives us 150, we timeout at 135

39 +

40 + time_t remaining_seconds = timeout - (time_t)(step_start_duration / USEC_PER_SEC);

41 + if(remaining_seconds < 0)

42 + remaining_seconds = 0;

43 +

44 + bool ok = completion_timedwait_for(&watcher_steps[step_id].p, remaining_seconds);

45 #else

46 // Wait indefinitely

47 bool ok = true;

@@ -37,16 +50,20 @@ static void watcher_wait_for_step(const watcher_step_id_t step_id)

51 usec_t step_duration = now_monotonic_usec() - step_start_time;

53 + char step_duration_txt[64];

54 + duration_snprintf(

55 + step_duration_txt, sizeof(step_duration_txt), (int64_t)(step_duration), "us", true);

56 +

57 if (ok) {

41 - netdata_log_info("shutdown step: [%d/%d] - '%s' finished in %llu milliseconds",

42 - (int)step_id + 1, (int)WATCHER_STEP_ID_MAX,

43 - watcher_steps[step_id].msg, step_duration / USEC_PER_MS);

58 + netdata_log_info("shutdown step: [%d/%d] - {at %s} finished '%s' in %s",

59 + (int)step_id + 1, (int)WATCHER_STEP_ID_MAX, start_duration_txt,

60 + watcher_steps[step_id].msg, step_duration_txt);

61 } else {

62 // Do not call fatal() because it will try to execute the exit

63 // sequence twice.

47 - netdata_log_error("shutdown step: [%d/%d] - '%s' took more than %u seconds (ie. %llu milliseconds)",

48 - (int)step_id + 1, (int)WATCHER_STEP_ID_MAX, watcher_steps[step_id].msg,

49 - timeout, step_duration / USEC_PER_MS);

64 + netdata_log_error("shutdown step: [%d/%d] - {at %s} timeout '%s' takes too long (%s) - giving up...",

65 + (int)step_id + 1, (int)WATCHER_STEP_ID_MAX, start_duration_txt,

66 + watcher_steps[step_id].msg, step_duration_txt);

68 abort();

69 }

@@ -64,32 +81,30 @@ void *watcher_main(void *arg)

82 usec_t shutdown_start_time = now_monotonic_usec();

67 - watcher_wait_for_step(WATCHER_STEP_ID_CREATE_SHUTDOWN_FILE);

68 - watcher_wait_for_step(WATCHER_STEP_ID_DESTROY_MAIN_SPAWN_SERVER);

69 - watcher_wait_for_step(WATCHER_STEP_ID_DBENGINE_EXIT_MODE);

70 - watcher_wait_for_step(WATCHER_STEP_ID_CLOSE_WEBRTC_CONNECTIONS);

71 - watcher_wait_for_step(WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS_AND_ACLK);

72 - watcher_wait_for_step(WATCHER_STEP_ID_STOP_MAINTENANCE_THREAD);

73 - watcher_wait_for_step(WATCHER_STEP_ID_STOP_EXPORTERS_HEALTH_AND_WEB_SERVERS_THREADS);

74 - watcher_wait_for_step(WATCHER_STEP_ID_STOP_COLLECTORS_AND_STREAMING_THREADS);

75 - watcher_wait_for_step(WATCHER_STEP_ID_STOP_REPLICATION_THREADS);

76 - watcher_wait_for_step(WATCHER_STEP_ID_PREPARE_METASYNC_SHUTDOWN);

77 - watcher_wait_for_step(WATCHER_STEP_ID_DISABLE_ML_DETECTION_AND_TRAINING_THREADS);

78 - watcher_wait_for_step(WATCHER_STEP_ID_STOP_CONTEXT_THREAD);

79 - watcher_wait_for_step(WATCHER_STEP_ID_CLEAR_WEB_CLIENT_CACHE);

80 - watcher_wait_for_step(WATCHER_STEP_ID_STOP_ACLK_THREADS);

81 - watcher_wait_for_step(WATCHER_STEP_ID_STOP_ALL_REMAINING_WORKER_THREADS);

82 - watcher_wait_for_step(WATCHER_STEP_ID_CANCEL_MAIN_THREADS);

83 - watcher_wait_for_step(WATCHER_STEP_ID_FLUSH_DBENGINE_TIERS);

84 - watcher_wait_for_step(WATCHER_STEP_ID_STOP_COLLECTION_FOR_ALL_HOSTS);

85 - watcher_wait_for_step(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_COLLECTORS_TO_FINISH);

86 - watcher_wait_for_step(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_MAIN_CACHE_TO_FINISH_FLUSHING);

87 - watcher_wait_for_step(WATCHER_STEP_ID_STOP_DBENGINE_TIERS);

88 - watcher_wait_for_step(WATCHER_STEP_ID_STOP_METASYNC_THREADS);

89 - watcher_wait_for_step(WATCHER_STEP_ID_CLOSE_SQL_DATABASES);

90 - watcher_wait_for_step(WATCHER_STEP_ID_REMOVE_PID_FILE);

91 - watcher_wait_for_step(WATCHER_STEP_ID_FREE_OPENSSL_STRUCTURES);

92 - watcher_wait_for_step(WATCHER_STEP_ID_REMOVE_INCOMPLETE_SHUTDOWN_FILE);

84 + watcher_wait_for_step(WATCHER_STEP_ID_CREATE_SHUTDOWN_FILE, shutdown_start_time);

85 + watcher_wait_for_step(WATCHER_STEP_ID_DESTROY_MAIN_SPAWN_SERVER, shutdown_start_time);

86 + watcher_wait_for_step(WATCHER_STEP_ID_DBENGINE_EXIT_MODE, shutdown_start_time);

87 + watcher_wait_for_step(WATCHER_STEP_ID_CLOSE_WEBRTC_CONNECTIONS, shutdown_start_time);

88 + watcher_wait_for_step(WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS_AND_ACLK, shutdown_start_time);

89 + watcher_wait_for_step(WATCHER_STEP_ID_STOP_MAINTENANCE_THREAD, shutdown_start_time);

90 + watcher_wait_for_step(WATCHER_STEP_ID_STOP_EXPORTERS_HEALTH_AND_WEB_SERVERS_THREADS, shutdown_start_time);

91 + watcher_wait_for_step(WATCHER_STEP_ID_STOP_COLLECTORS_AND_STREAMING_THREADS, shutdown_start_time);

92 + watcher_wait_for_step(WATCHER_STEP_ID_STOP_REPLICATION_THREADS, shutdown_start_time);

93 + watcher_wait_for_step(WATCHER_STEP_ID_DISABLE_ML_DETECTION_AND_TRAINING_THREADS, shutdown_start_time);

94 + watcher_wait_for_step(WATCHER_STEP_ID_STOP_CONTEXT_THREAD, shutdown_start_time);

95 + watcher_wait_for_step(WATCHER_STEP_ID_CLEAR_WEB_CLIENT_CACHE, shutdown_start_time);

96 + watcher_wait_for_step(WATCHER_STEP_ID_STOP_ACLK_THREADS, shutdown_start_time);

97 + watcher_wait_for_step(WATCHER_STEP_ID_STOP_ALL_REMAINING_WORKER_THREADS, shutdown_start_time);

98 + watcher_wait_for_step(WATCHER_STEP_ID_CANCEL_MAIN_THREADS, shutdown_start_time);

99 + watcher_wait_for_step(WATCHER_STEP_ID_PREPARE_METASYNC_SHUTDOWN, shutdown_start_time);

100 + watcher_wait_for_step(WATCHER_STEP_ID_STOP_COLLECTION_FOR_ALL_HOSTS, shutdown_start_time);

101 + watcher_wait_for_step(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_COLLECTORS_TO_FINISH, shutdown_start_time);

102 + watcher_wait_for_step(WATCHER_STEP_ID_STOP_DBENGINE_TIERS, shutdown_start_time);

103 + watcher_wait_for_step(WATCHER_STEP_ID_STOP_METASYNC_THREADS, shutdown_start_time);

104 + watcher_wait_for_step(WATCHER_STEP_ID_CLOSE_SQL_DATABASES, shutdown_start_time);

105 + watcher_wait_for_step(WATCHER_STEP_ID_REMOVE_PID_FILE, shutdown_start_time);

106 + watcher_wait_for_step(WATCHER_STEP_ID_FREE_OPENSSL_STRUCTURES, shutdown_start_time);

107 + watcher_wait_for_step(WATCHER_STEP_ID_REMOVE_INCOMPLETE_SHUTDOWN_FILE, shutdown_start_time);

108

109 completion_wait_for(&shutdown_end_completion);

110 usec_t shutdown_end_time = now_monotonic_usec();

@@ -136,14 +151,10 @@ void watcher_thread_start() {

151 "stop all remaining worker threads";

152 watcher_steps[WATCHER_STEP_ID_CANCEL_MAIN_THREADS].msg =

153 "cancel main threads";

139 - watcher_steps[WATCHER_STEP_ID_FLUSH_DBENGINE_TIERS].msg =

140 - "flush dbengine tiers";

154 watcher_steps[WATCHER_STEP_ID_STOP_COLLECTION_FOR_ALL_HOSTS].msg =

155 "stop collection for all hosts";

156 watcher_steps[WATCHER_STEP_ID_WAIT_FOR_DBENGINE_COLLECTORS_TO_FINISH].msg =

157 "wait for dbengine collectors to finish";

145 - watcher_steps[WATCHER_STEP_ID_WAIT_FOR_DBENGINE_MAIN_CACHE_TO_FINISH_FLUSHING].msg =

146 - "wait for dbengine main cache to finish flushing";

158 watcher_steps[WATCHER_STEP_ID_STOP_DBENGINE_TIERS].msg =

159 "stop dbengine tiers";

160 watcher_steps[WATCHER_STEP_ID_STOP_METASYNC_THREADS].msg =

src/daemon/daemon-shutdown-watcher.h renamed

+1 -3

@@ -15,17 +15,15 @@ typedef enum {

15 WATCHER_STEP_ID_STOP_EXPORTERS_HEALTH_AND_WEB_SERVERS_THREADS,

16 WATCHER_STEP_ID_STOP_COLLECTORS_AND_STREAMING_THREADS,

17 WATCHER_STEP_ID_STOP_REPLICATION_THREADS,

18 - WATCHER_STEP_ID_PREPARE_METASYNC_SHUTDOWN,

18 WATCHER_STEP_ID_DISABLE_ML_DETECTION_AND_TRAINING_THREADS,

19 WATCHER_STEP_ID_STOP_CONTEXT_THREAD,

20 WATCHER_STEP_ID_CLEAR_WEB_CLIENT_CACHE,

21 WATCHER_STEP_ID_STOP_ACLK_THREADS,

22 WATCHER_STEP_ID_STOP_ALL_REMAINING_WORKER_THREADS,

23 WATCHER_STEP_ID_CANCEL_MAIN_THREADS,

25 - WATCHER_STEP_ID_FLUSH_DBENGINE_TIERS,

24 + WATCHER_STEP_ID_PREPARE_METASYNC_SHUTDOWN,

25 WATCHER_STEP_ID_STOP_COLLECTION_FOR_ALL_HOSTS,

26 WATCHER_STEP_ID_WAIT_FOR_DBENGINE_COLLECTORS_TO_FINISH,

28 - WATCHER_STEP_ID_WAIT_FOR_DBENGINE_MAIN_CACHE_TO_FINISH_FLUSHING,

27 WATCHER_STEP_ID_STOP_DBENGINE_TIERS,

28 WATCHER_STEP_ID_STOP_METASYNC_THREADS,

29 WATCHER_STEP_ID_CLOSE_SQL_DATABASES,

src/daemon/daemon-shutdown.c new

+301

@@ -0,0 +1,301 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#include "daemon-shutdown.h"

4 +#include "daemon-service.h"

5 +#include "daemon/daemon-shutdown-watcher.h"

6 +#include "static_threads.h"

7 +#include "common.h"

8 +

9 +#include <curl/curl.h>

10 +

11 +#ifdef ENABLE_SENTRY

12 +#include "sentry-native/sentry-native.h"

13 +#endif

14 +

15 +void web_client_cache_destroy(void);

16 +

17 +extern struct netdata_static_thread *static_threads;

18 +

19 +void cancel_main_threads(void) {

20 + nd_log_limits_unlimited();

21 +

22 + if (!static_threads)

23 + return;

24 +

25 + int i, found = 0;

26 + usec_t max = 5 * USEC_PER_SEC, step = 100000;

27 + for (i = 0; static_threads[i].name != NULL ; i++) {

28 + if (static_threads[i].enabled == NETDATA_MAIN_THREAD_RUNNING) {

29 + if (static_threads[i].thread) {

30 + netdata_log_info("EXIT: Stopping main thread: %s", static_threads[i].name);

31 + nd_thread_signal_cancel(static_threads[i].thread);

32 + } else {

33 + netdata_log_info("EXIT: No thread running (marking as EXITED): %s", static_threads[i].name);

34 + static_threads[i].enabled = NETDATA_MAIN_THREAD_EXITED;

35 + }

36 + found++;

37 + }

38 + }

39 +

40 + while(found && max > 0) {

41 + max -= step;

42 + netdata_log_info("Waiting %d threads to finish...", found);

43 + sleep_usec(step);

44 + found = 0;

45 + for (i = 0; static_threads[i].name != NULL ; i++) {

46 + if (static_threads[i].enabled == NETDATA_MAIN_THREAD_EXITED)

47 + continue;

48 +

49 + // Don't wait ourselves.

50 + if (nd_thread_is_me(static_threads[i].thread))

51 + continue;

52 +

53 + found++;

54 + }

55 + }

56 +

57 + if(found) {

58 + for (i = 0; static_threads[i].name != NULL ; i++) {

59 + if (static_threads[i].enabled != NETDATA_MAIN_THREAD_EXITED)

60 + netdata_log_error("Main thread %s takes too long to exit. Giving up...", static_threads[i].name);

61 + }

62 + }

63 + else

64 + netdata_log_info("All threads finished.");

65 +

66 + freez(static_threads);

67 + static_threads = NULL;

68 +}

69 +

70 +static void *rrdeng_exit_background(void *ptr) {

71 + struct rrdengine_instance *ctx = ptr;

72 + rrdeng_exit(ctx);

73 + return NULL;

74 +}

75 +

76 +#ifdef ENABLE_DBENGINE

77 +static void rrdeng_flush_everything_and_wait(bool wait_flush, bool wait_collectors) {

78 + static size_t starting_size_to_flush = 0;

79 +

80 + if(!pgc_hot_and_dirty_entries(main_cache))

81 + return;

82 +

83 + nd_log(NDLS_DAEMON, NDLP_INFO, "Flushing DBENGINE dirty pages...");

84 + for (size_t tier = 0; tier < storage_tiers; tier++)

85 + rrdeng_quiesce(multidb_ctx[tier]);

86 +

87 + struct pgc_statistics pgc_main_stats = pgc_get_statistics(main_cache);

88 + size_t size_to_flush = pgc_main_stats.queues[PGC_QUEUE_HOT].size + pgc_main_stats.queues[PGC_QUEUE_DIRTY].size;

89 + if(size_to_flush > starting_size_to_flush || !starting_size_to_flush)

90 + starting_size_to_flush = size_to_flush;

91 +

92 + if(wait_collectors) {

93 + size_t running = 1;

94 + size_t count = 10;

95 + while (running && count) {

96 + running = 0;

97 + for (size_t tier = 0; tier < storage_tiers; tier++)

98 + running += rrdeng_collectors_running(multidb_ctx[tier]);

99 +

100 + if (running) {

101 + nd_log_limit_static_thread_var(erl, 1, 100 * USEC_PER_MS);

102 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_NOTICE, "waiting for %zu collectors to finish", running);

103 + }

104 + count--;

105 + }

106 + }

107 +

108 + if(!wait_flush)

109 + return;

110 +

111 + for(size_t iterations = 0; true ;iterations++) {

112 + pgc_main_stats = pgc_get_statistics(main_cache);

113 + size_to_flush = pgc_main_stats.queues[PGC_QUEUE_HOT].size + pgc_main_stats.queues[PGC_QUEUE_DIRTY].size;

114 + if(!starting_size_to_flush || size_to_flush > starting_size_to_flush)

115 + starting_size_to_flush = size_to_flush;

116 +

117 + if(!size_to_flush)

118 + break;

119 +

120 + size_t flushed = starting_size_to_flush - size_to_flush;

121 +

122 + if(iterations % 10 == 0) {

123 + char hot[64], dirty[64];

124 + size_snprintf(hot, sizeof(hot), pgc_main_stats.queues[PGC_QUEUE_HOT].size, "B", false);

125 + size_snprintf(dirty, sizeof(hot), pgc_main_stats.queues[PGC_QUEUE_DIRTY].size, "B", false);

126 +

127 + nd_log(NDLS_DAEMON, NDLP_INFO, "DBENGINE: flushing at %.2f%% { hot: %s, dirty: %s }...",

128 + (double)flushed * 100.0 / (double)starting_size_to_flush,

129 + hot, dirty);

130 + }

131 + sleep_usec(100 * USEC_PER_MS);

132 + }

133 + nd_log(NDLS_DAEMON, NDLP_INFO, "DBENGINE: flushing completed!");

134 +}

135 +#endif

136 +

137 +void netdata_cleanup_and_exit(int ret, const char *action, const char *action_result, const char *action_data) {

138 + netdata_exit = 1;

139 +

140 +#ifdef ENABLE_DBENGINE

141 + if(!ret && dbengine_enabled)

142 + // flush all dirty pages asap

143 + rrdeng_flush_everything_and_wait(false, false);

144 +#endif

145 +

146 + usec_t shutdown_start_time = now_monotonic_usec();

147 + watcher_shutdown_begin();

148 +

149 + nd_log_limits_unlimited();

150 + netdata_log_info("NETDATA SHUTDOWN: initializing shutdown with code %d...", ret);

151 +

152 + // send the stat from our caller

153 + analytics_statistic_t statistic = { action, action_result, action_data };

154 + analytics_statistic_send(&statistic);

155 +

156 + // notify we are exiting

157 + statistic = (analytics_statistic_t) {"EXIT", ret?"ERROR":"OK","-"};

158 + analytics_statistic_send(&statistic);

159 +

160 + char agent_crash_file[FILENAME_MAX + 1];

161 + char agent_incomplete_shutdown_file[FILENAME_MAX + 1];

162 + snprintfz(agent_crash_file, FILENAME_MAX, "%s/.agent_crash", netdata_configured_varlib_dir);

163 + snprintfz(agent_incomplete_shutdown_file, FILENAME_MAX, "%s/.agent_incomplete_shutdown", netdata_configured_varlib_dir);

164 + (void) rename(agent_crash_file, agent_incomplete_shutdown_file);

165 + watcher_step_complete(WATCHER_STEP_ID_CREATE_SHUTDOWN_FILE);

166 +

167 + netdata_main_spawn_server_cleanup();

168 + watcher_step_complete(WATCHER_STEP_ID_DESTROY_MAIN_SPAWN_SERVER);

169 +

170 + watcher_step_complete(WATCHER_STEP_ID_DBENGINE_EXIT_MODE);

171 +

172 + webrtc_close_all_connections();

173 + watcher_step_complete(WATCHER_STEP_ID_CLOSE_WEBRTC_CONNECTIONS);

174 +

175 + service_signal_exit(SERVICE_MAINTENANCE | ABILITY_DATA_QUERIES | ABILITY_WEB_REQUESTS |

176 + ABILITY_STREAMING_CONNECTIONS | SERVICE_ACLK);

177 + watcher_step_complete(WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS_AND_ACLK);

178 +

179 + service_wait_exit(SERVICE_MAINTENANCE, 3 * USEC_PER_SEC);

180 + watcher_step_complete(WATCHER_STEP_ID_STOP_MAINTENANCE_THREAD);

181 +

182 + service_wait_exit(SERVICE_EXPORTERS | SERVICE_HEALTH | SERVICE_WEB_SERVER | SERVICE_HTTPD, 3 * USEC_PER_SEC);

183 + watcher_step_complete(WATCHER_STEP_ID_STOP_EXPORTERS_HEALTH_AND_WEB_SERVERS_THREADS);

184 +

185 + stream_threads_cancel();

186 + service_wait_exit(SERVICE_COLLECTORS | SERVICE_STREAMING, 3 * USEC_PER_SEC);

187 + watcher_step_complete(WATCHER_STEP_ID_STOP_COLLECTORS_AND_STREAMING_THREADS);

188 +

189 +#ifdef ENABLE_DBENGINE

190 + if(!ret && dbengine_enabled)

191 + // flush all dirty pages now that all collectors and streaming completed

192 + rrdeng_flush_everything_and_wait(false, false);

193 +#endif

194 +

195 + service_wait_exit(SERVICE_REPLICATION, 3 * USEC_PER_SEC);

196 + watcher_step_complete(WATCHER_STEP_ID_STOP_REPLICATION_THREADS);

197 +

198 + ml_stop_threads();

199 + ml_fini();

200 + watcher_step_complete(WATCHER_STEP_ID_DISABLE_ML_DETECTION_AND_TRAINING_THREADS);

201 +

202 + service_wait_exit(SERVICE_CONTEXT, 3 * USEC_PER_SEC);

203 + watcher_step_complete(WATCHER_STEP_ID_STOP_CONTEXT_THREAD);

204 +

205 + web_client_cache_destroy();

206 + watcher_step_complete(WATCHER_STEP_ID_CLEAR_WEB_CLIENT_CACHE);

207 +

208 + service_wait_exit(SERVICE_ACLK, 3 * USEC_PER_SEC);

209 + watcher_step_complete(WATCHER_STEP_ID_STOP_ACLK_THREADS);

210 +

211 + service_wait_exit(~0, 10 * USEC_PER_SEC);

212 + watcher_step_complete(WATCHER_STEP_ID_STOP_ALL_REMAINING_WORKER_THREADS);

213 +

214 + cancel_main_threads();

215 + watcher_step_complete(WATCHER_STEP_ID_CANCEL_MAIN_THREADS);

216 +

217 + metadata_sync_shutdown_background();

218 + watcher_step_complete(WATCHER_STEP_ID_PREPARE_METASYNC_SHUTDOWN);

219 +

220 + if (ret)

221 + {

222 + watcher_step_complete(WATCHER_STEP_ID_STOP_COLLECTION_FOR_ALL_HOSTS);

223 + watcher_step_complete(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_COLLECTORS_TO_FINISH);

224 + watcher_step_complete(WATCHER_STEP_ID_STOP_DBENGINE_TIERS);

225 + watcher_step_complete(WATCHER_STEP_ID_STOP_METASYNC_THREADS);

226 + }

227 + else

228 + {

229 + // exit cleanly

230 + rrd_finalize_collection_for_all_hosts();

231 + watcher_step_complete(WATCHER_STEP_ID_STOP_COLLECTION_FOR_ALL_HOSTS);

232 +

233 +#ifdef ENABLE_DBENGINE

234 + if(dbengine_enabled) {

235 + // flush anything remaining and wait for collectors to finish

236 + rrdeng_flush_everything_and_wait(true, true);

237 + watcher_step_complete(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_COLLECTORS_TO_FINISH);

238 +

239 + ND_THREAD *th[storage_tiers];

240 + for (size_t tier = 0; tier < storage_tiers; tier++)

241 + th[tier] = nd_thread_create("rrdeng-exit", NETDATA_THREAD_OPTION_JOINABLE, rrdeng_exit_background, multidb_ctx[tier]);

242 +

243 + // flush anything remaining again - just in case

244 + rrdeng_flush_everything_and_wait(true, false);

245 +

246 + for (size_t tier = 0; tier < storage_tiers; tier++)

247 + nd_thread_join(th[tier]);

248 +

249 + rrdeng_enq_cmd(NULL, RRDENG_OPCODE_SHUTDOWN_EVLOOP, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

250 + watcher_step_complete(WATCHER_STEP_ID_STOP_DBENGINE_TIERS);

251 + }

252 + else {

253 + // Skip these steps

254 + watcher_step_complete(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_COLLECTORS_TO_FINISH);

255 + watcher_step_complete(WATCHER_STEP_ID_STOP_DBENGINE_TIERS);

256 + }

257 +#else

258 + // Skip these steps

259 + watcher_step_complete(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_COLLECTORS_TO_FINISH);

260 + watcher_step_complete(WATCHER_STEP_ID_STOP_DBENGINE_TIERS);

261 +#endif

262 +

263 + metadata_sync_shutdown_background_wait();

264 + watcher_step_complete(WATCHER_STEP_ID_STOP_METASYNC_THREADS);

265 + }

266 +

267 + // Don't register a shutdown event if we crashed

268 + if (!ret)

269 + add_agent_event(EVENT_AGENT_SHUTDOWN_TIME, (int64_t)(now_monotonic_usec() - shutdown_start_time));

270 + sqlite_close_databases();

271 + watcher_step_complete(WATCHER_STEP_ID_CLOSE_SQL_DATABASES);

272 + sqlite_library_shutdown();

273 +

274 +

275 + // unlink the pid

276 + if(pidfile && *pidfile) {

277 + if(unlink(pidfile) != 0)

278 + netdata_log_error("EXIT: cannot unlink pidfile '%s'.", pidfile);

279 + }

280 + watcher_step_complete(WATCHER_STEP_ID_REMOVE_PID_FILE);

281 +

282 + netdata_ssl_cleanup();

283 + watcher_step_complete(WATCHER_STEP_ID_FREE_OPENSSL_STRUCTURES);

284 +

285 + (void) unlink(agent_incomplete_shutdown_file);

286 + watcher_step_complete(WATCHER_STEP_ID_REMOVE_INCOMPLETE_SHUTDOWN_FILE);

287 +

288 + watcher_shutdown_end();

289 + watcher_thread_stop();

290 + curl_global_cleanup();

291 +

292 +#ifdef OS_WINDOWS

293 + return;

294 +#endif

295 +

296 +#ifdef ENABLE_SENTRY

297 + nd_sentry_fini();

298 +#endif

299 +

300 + exit(ret);

301 +}

src/daemon/daemon-shutdown.h new

+10

@@ -0,0 +1,10 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#ifndef NETDATA_DAEMON_SHUTDOWN_H

4 +#define NETDATA_DAEMON_SHUTDOWN_H

5 +

6 +#include "libnetdata/libnetdata.h"

7 +

8 +void cancel_main_threads(void);

9 +

10 +#endif //NETDATA_DAEMON_SHUTDOWN_H

src/daemon/main.c

+2 -556

@@ -2,7 +2,7 @@

3 #include "common.h"

4 #include "buildinfo.h"

5 -#include "daemon/watcher.h"

5 +#include "daemon/daemon-shutdown-watcher.h"

6 #include "static_threads.h"

7 #include "web/api/queries/backfill.h"

@@ -29,509 +29,6 @@ bool ieee754_doubles = false;

29 time_t netdata_start_time = 0;

30 struct netdata_static_thread *static_threads;

32 -struct config netdata_config = APPCONFIG_INITIALIZER;

33 -

34 -typedef struct service_thread {

35 - pid_t tid;

36 - SERVICE_THREAD_TYPE type;

37 - SERVICE_TYPE services;

38 - char name[ND_THREAD_TAG_MAX + 1];

39 - bool stop_immediately;

40 - bool cancelled;

41 -

42 - union {

43 - ND_THREAD *netdata_thread;

44 - uv_thread_t uv_thread;

45 - };

46 -

47 - force_quit_t force_quit_callback;

48 - request_quit_t request_quit_callback;

49 - void *data;

50 -} SERVICE_THREAD;

51 -

52 -struct service_globals {

53 - SPINLOCK lock;

54 - Pvoid_t pid_judy;

55 -} service_globals = {

56 - .pid_judy = NULL,

57 -};

58 -

59 -SERVICE_THREAD *service_register(SERVICE_THREAD_TYPE thread_type, request_quit_t request_quit_callback, force_quit_t force_quit_callback, void *data, bool update __maybe_unused) {

60 - SERVICE_THREAD *sth = NULL;

61 - pid_t tid = gettid_cached();

62 -

63 - spinlock_lock(&service_globals.lock);

64 - Pvoid_t *PValue = JudyLIns(&service_globals.pid_judy, tid, PJE0);

65 - if(!*PValue) {

66 - sth = callocz(1, sizeof(SERVICE_THREAD));

67 - sth->tid = tid;

68 - sth->type = thread_type;

69 - sth->request_quit_callback = request_quit_callback;

70 - sth->force_quit_callback = force_quit_callback;

71 - sth->data = data;

72 - *PValue = sth;

73 -

74 - switch(thread_type) {

75 - default:

76 - case SERVICE_THREAD_TYPE_NETDATA:

77 - sth->netdata_thread = nd_thread_self();

78 - break;

79 -

80 - case SERVICE_THREAD_TYPE_EVENT_LOOP:

81 - case SERVICE_THREAD_TYPE_LIBUV:

82 - sth->uv_thread = uv_thread_self();

83 - break;

84 - }

85 -

86 - const char *name = nd_thread_tag();

87 - if(!name) name = "";

88 - strncpyz(sth->name, name, sizeof(sth->name) - 1);

89 - }

90 - else {

91 - sth = *PValue;

92 - }

93 - spinlock_unlock(&service_globals.lock);

94 -

95 - return sth;

96 -}

97 -

98 -void service_exits(void) {

99 - pid_t tid = gettid_cached();

100 -

101 - spinlock_lock(&service_globals.lock);

102 - Pvoid_t *PValue = JudyLGet(service_globals.pid_judy, tid, PJE0);

103 - if(PValue) {

104 - freez(*PValue);

105 - JudyLDel(&service_globals.pid_judy, tid, PJE0);

106 - }

107 - spinlock_unlock(&service_globals.lock);

108 -}

109 -

110 -bool service_running(SERVICE_TYPE service) {

111 - static __thread SERVICE_THREAD *sth = NULL;

112 -

113 - if(unlikely(!sth))

114 - sth = service_register(SERVICE_THREAD_TYPE_NETDATA, NULL, NULL, NULL, false);

115 -

116 - sth->services |= service;

117 -

118 - bool cancelled = false;

119 - if (sth->type == SERVICE_THREAD_TYPE_NETDATA)

120 - cancelled = nd_thread_signaled_to_cancel();

121 -

122 - return !sth->stop_immediately && !netdata_exit && !cancelled;

123 -}

124 -

125 -void service_signal_exit(SERVICE_TYPE service) {

126 - spinlock_lock(&service_globals.lock);

127 -

128 - Pvoid_t *PValue;

129 - Word_t tid = 0;

130 - bool first = true;

131 - while((PValue = JudyLFirstThenNext(service_globals.pid_judy, &tid, &first))) {

132 - SERVICE_THREAD *sth = *PValue;

133 -

134 - if((sth->services & service)) {

135 - sth->stop_immediately = true;

136 -

137 - switch(sth->type) {

138 - default:

139 - case SERVICE_THREAD_TYPE_NETDATA:

140 - nd_thread_signal_cancel(sth->netdata_thread);

141 - break;

142 -

143 - case SERVICE_THREAD_TYPE_EVENT_LOOP:

144 - case SERVICE_THREAD_TYPE_LIBUV:

145 - break;

146 - }

147 -

148 - if(sth->request_quit_callback) {

149 - spinlock_unlock(&service_globals.lock);

150 - sth->request_quit_callback(sth->data);

151 - spinlock_lock(&service_globals.lock);

152 - }

153 - }

154 - }

155 -

156 - spinlock_unlock(&service_globals.lock);

157 -}

158 -

159 -static void service_to_buffer(BUFFER *wb, SERVICE_TYPE service) {

160 - if(service & SERVICE_MAINTENANCE)

161 - buffer_strcat(wb, "MAINTENANCE ");

162 - if(service & SERVICE_COLLECTORS)

163 - buffer_strcat(wb, "COLLECTORS ");

164 - if(service & SERVICE_REPLICATION)

165 - buffer_strcat(wb, "REPLICATION ");

166 - if(service & ABILITY_DATA_QUERIES)

167 - buffer_strcat(wb, "DATA_QUERIES ");

168 - if(service & ABILITY_WEB_REQUESTS)

169 - buffer_strcat(wb, "WEB_REQUESTS ");

170 - if(service & SERVICE_WEB_SERVER)

171 - buffer_strcat(wb, "WEB_SERVER ");

172 - if(service & SERVICE_ACLK)

173 - buffer_strcat(wb, "ACLK ");

174 - if(service & SERVICE_HEALTH)

175 - buffer_strcat(wb, "HEALTH ");

176 - if(service & SERVICE_STREAMING)

177 - buffer_strcat(wb, "STREAMING ");

178 - if(service & ABILITY_STREAMING_CONNECTIONS)

179 - buffer_strcat(wb, "STREAMING_CONNECTIONS ");

180 - if(service & SERVICE_CONTEXT)

181 - buffer_strcat(wb, "CONTEXT ");

182 - if(service & SERVICE_ANALYTICS)

183 - buffer_strcat(wb, "ANALYTICS ");

184 - if(service & SERVICE_EXPORTERS)

185 - buffer_strcat(wb, "EXPORTERS ");

186 - if(service & SERVICE_HTTPD)

187 - buffer_strcat(wb, "HTTPD ");

188 -}

189 -

190 -static bool service_wait_exit(SERVICE_TYPE service, usec_t timeout_ut) {

191 - BUFFER *service_list = buffer_create(1024, NULL);

192 - BUFFER *thread_list = buffer_create(1024, NULL);

193 - usec_t started_ut = now_monotonic_usec(), ended_ut;

194 - size_t running;

195 - SERVICE_TYPE running_services = 0;

196 -

197 - // cancel the threads

198 - running = 0;

199 - running_services = 0;

200 - {

201 - buffer_flush(thread_list);

202 -

203 - spinlock_lock(&service_globals.lock);

204 -

205 - Pvoid_t *PValue;

206 - Word_t tid = 0;

207 - bool first = true;

208 - while((PValue = JudyLFirstThenNext(service_globals.pid_judy, &tid, &first))) {

209 - SERVICE_THREAD *sth = *PValue;

210 - if(sth->services & service && sth->tid != gettid_cached() && !sth->cancelled) {

211 - sth->cancelled = true;

212 -

213 - switch(sth->type) {

214 - default:

215 - case SERVICE_THREAD_TYPE_NETDATA:

216 - nd_thread_signal_cancel(sth->netdata_thread);

217 - break;

218 -

219 - case SERVICE_THREAD_TYPE_EVENT_LOOP:

220 - case SERVICE_THREAD_TYPE_LIBUV:

221 - break;

222 - }

223 -

224 - if(running)

225 - buffer_strcat(thread_list, ", ");

226 -

227 - buffer_sprintf(thread_list, "'%s' (%d)", sth->name, sth->tid);

228 -

229 - running++;

230 - running_services |= sth->services & service;

231 -

232 - if(sth->force_quit_callback) {

233 - spinlock_unlock(&service_globals.lock);

234 - sth->force_quit_callback(sth->data);

235 - spinlock_lock(&service_globals.lock);

236 - continue;

237 - }

238 - }

239 - }

240 -

241 - spinlock_unlock(&service_globals.lock);

242 - }

243 -

244 - service_signal_exit(service);

245 -

246 - // signal them to stop

247 - size_t last_running = 0;

248 - size_t stale_time_ut = 0;

249 - usec_t sleep_ut = 50 * USEC_PER_MS;

250 - size_t log_countdown_ut = sleep_ut;

251 - do {

252 - if(running != last_running)

253 - stale_time_ut = 0;

254 -

255 - last_running = running;

256 - running = 0;

257 - running_services = 0;

258 - buffer_flush(thread_list);

259 -

260 - spinlock_lock(&service_globals.lock);

261 -

262 - Pvoid_t *PValue;

263 - Word_t tid = 0;

264 - bool first = true;

265 - while((PValue = JudyLFirstThenNext(service_globals.pid_judy, &tid, &first))) {

266 - SERVICE_THREAD *sth = *PValue;

267 - if(sth->services & service && sth->tid != gettid_cached()) {

268 - if(running)

269 - buffer_strcat(thread_list, ", ");

270 -

271 - buffer_sprintf(thread_list, "'%s' (%d)", sth->name, sth->tid);

272 -

273 - running_services |= sth->services & service;

274 - running++;

275 - }

276 - }

277 -

278 - spinlock_unlock(&service_globals.lock);

279 -

280 - if(running) {

281 - log_countdown_ut -= (log_countdown_ut >= sleep_ut) ? sleep_ut : log_countdown_ut;

282 - if(log_countdown_ut == 0 || running != last_running) {

283 - log_countdown_ut = 20 * sleep_ut;

284 -

285 - buffer_flush(service_list);

286 - service_to_buffer(service_list, running_services);

287 - netdata_log_info("SERVICE CONTROL: waiting for the following %zu services [ %s] to exit: %s",

288 - running, buffer_tostring(service_list),

289 - running <= 10 ? buffer_tostring(thread_list) : "");

290 - }

291 -

292 - sleep_usec(sleep_ut);

293 - stale_time_ut += sleep_ut;

294 - }

295 -

296 - ended_ut = now_monotonic_usec();

297 - } while(running && (ended_ut - started_ut < timeout_ut || stale_time_ut < timeout_ut));

298 -

299 - if(running) {

300 - buffer_flush(service_list);

301 - service_to_buffer(service_list, running_services);

302 - netdata_log_info("SERVICE CONTROL: "

303 - "the following %zu service(s) [ %s] take too long to exit: %s; "

304 - "giving up on them...",

305 - running, buffer_tostring(service_list),

306 - buffer_tostring(thread_list));

307 - }

308 -

309 - buffer_free(thread_list);

310 - buffer_free(service_list);

311 -

312 - return (running == 0);

313 -}

314 -

315 -void web_client_cache_destroy(void);

316 -

317 -void netdata_cleanup_and_exit(int ret, const char *action, const char *action_result, const char *action_data) {

318 - netdata_exit = 1;

319 -

320 - usec_t shutdown_start_time = now_monotonic_usec();

321 - watcher_shutdown_begin();

322 -

323 - nd_log_limits_unlimited();

324 - netdata_log_info("NETDATA SHUTDOWN: initializing shutdown with code %d...", ret);

325 -

326 - // send the stat from our caller

327 - analytics_statistic_t statistic = { action, action_result, action_data };

328 - analytics_statistic_send(&statistic);

329 -

330 - // notify we are exiting

331 - statistic = (analytics_statistic_t) {"EXIT", ret?"ERROR":"OK","-"};

332 - analytics_statistic_send(&statistic);

333 -

334 - char agent_crash_file[FILENAME_MAX + 1];

335 - char agent_incomplete_shutdown_file[FILENAME_MAX + 1];

336 - snprintfz(agent_crash_file, FILENAME_MAX, "%s/.agent_crash", netdata_configured_varlib_dir);

337 - snprintfz(agent_incomplete_shutdown_file, FILENAME_MAX, "%s/.agent_incomplete_shutdown", netdata_configured_varlib_dir);

338 - (void) rename(agent_crash_file, agent_incomplete_shutdown_file);

339 - watcher_step_complete(WATCHER_STEP_ID_CREATE_SHUTDOWN_FILE);

340 -

341 - netdata_main_spawn_server_cleanup();

342 - watcher_step_complete(WATCHER_STEP_ID_DESTROY_MAIN_SPAWN_SERVER);

343 -

344 -#ifdef ENABLE_DBENGINE

345 - if(dbengine_enabled) {

346 - for (size_t tier = 0; tier < storage_tiers; tier++)

347 - rrdeng_exit_mode(multidb_ctx[tier]);

348 - }

349 -#endif

350 - watcher_step_complete(WATCHER_STEP_ID_DBENGINE_EXIT_MODE);

351 -

352 - webrtc_close_all_connections();

353 - watcher_step_complete(WATCHER_STEP_ID_CLOSE_WEBRTC_CONNECTIONS);

354 -

355 - service_signal_exit(SERVICE_MAINTENANCE | ABILITY_DATA_QUERIES | ABILITY_WEB_REQUESTS |

356 - ABILITY_STREAMING_CONNECTIONS | SERVICE_ACLK);

357 - watcher_step_complete(WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS_AND_ACLK);

358 -

359 - service_wait_exit(SERVICE_MAINTENANCE, 3 * USEC_PER_SEC);

360 - watcher_step_complete(WATCHER_STEP_ID_STOP_MAINTENANCE_THREAD);

361 -

362 - service_wait_exit(SERVICE_EXPORTERS | SERVICE_HEALTH | SERVICE_WEB_SERVER | SERVICE_HTTPD, 3 * USEC_PER_SEC);

363 - watcher_step_complete(WATCHER_STEP_ID_STOP_EXPORTERS_HEALTH_AND_WEB_SERVERS_THREADS);

364 -

365 - stream_threads_cancel();

366 - service_wait_exit(SERVICE_COLLECTORS | SERVICE_STREAMING, 3 * USEC_PER_SEC);

367 - watcher_step_complete(WATCHER_STEP_ID_STOP_COLLECTORS_AND_STREAMING_THREADS);

368 -

369 - service_wait_exit(SERVICE_REPLICATION, 3 * USEC_PER_SEC);

370 - watcher_step_complete(WATCHER_STEP_ID_STOP_REPLICATION_THREADS);

371 -

372 - metadata_sync_shutdown_prepare();

373 - watcher_step_complete(WATCHER_STEP_ID_PREPARE_METASYNC_SHUTDOWN);

374 -

375 - ml_stop_threads();

376 - ml_fini();

377 - watcher_step_complete(WATCHER_STEP_ID_DISABLE_ML_DETECTION_AND_TRAINING_THREADS);

378 -

379 - service_wait_exit(SERVICE_CONTEXT, 3 * USEC_PER_SEC);

380 - watcher_step_complete(WATCHER_STEP_ID_STOP_CONTEXT_THREAD);

381 -

382 - web_client_cache_destroy();

383 - watcher_step_complete(WATCHER_STEP_ID_CLEAR_WEB_CLIENT_CACHE);

384 -

385 - service_wait_exit(SERVICE_ACLK, 3 * USEC_PER_SEC);

386 - watcher_step_complete(WATCHER_STEP_ID_STOP_ACLK_THREADS);

387 -

388 - service_wait_exit(~0, 10 * USEC_PER_SEC);

389 - watcher_step_complete(WATCHER_STEP_ID_STOP_ALL_REMAINING_WORKER_THREADS);

390 -

391 - cancel_main_threads();

392 - watcher_step_complete(WATCHER_STEP_ID_CANCEL_MAIN_THREADS);

393 -

394 - if (ret)

395 - {

396 - watcher_step_complete(WATCHER_STEP_ID_FLUSH_DBENGINE_TIERS);

397 - watcher_step_complete(WATCHER_STEP_ID_STOP_COLLECTION_FOR_ALL_HOSTS);

398 - watcher_step_complete(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_COLLECTORS_TO_FINISH);

399 - watcher_step_complete(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_MAIN_CACHE_TO_FINISH_FLUSHING);

400 - watcher_step_complete(WATCHER_STEP_ID_STOP_DBENGINE_TIERS);

401 - watcher_step_complete(WATCHER_STEP_ID_STOP_METASYNC_THREADS);

402 - }

403 - else

404 - {

405 - // exit cleanly

406 -

407 -#ifdef ENABLE_DBENGINE

408 - if(dbengine_enabled) {

409 - nd_log(NDLS_DAEMON, NDLP_INFO, "Preparing DBENGINE shutdown...");

410 - for (size_t tier = 0; tier < storage_tiers; tier++)

411 - rrdeng_prepare_exit(multidb_ctx[tier]);

412 -

413 - struct pgc_statistics pgc_main_stats = pgc_get_statistics(main_cache);

414 - nd_log(NDLS_DAEMON, NDLP_INFO, "Waiting for DBENGINE to commit unsaved data to disk (%zu pages, %zu bytes)...",

415 - pgc_main_stats.queues[PGC_QUEUE_HOT].entries + pgc_main_stats.queues[PGC_QUEUE_DIRTY].entries,

416 - pgc_main_stats.queues[PGC_QUEUE_HOT].size + pgc_main_stats.queues[PGC_QUEUE_DIRTY].size);

417 -

418 - bool finished_tiers[RRD_STORAGE_TIERS] = { 0 };

419 - size_t waiting_tiers, iterations = 0;

420 - do {

421 - waiting_tiers = 0;

422 - iterations++;

423 -

424 - for (size_t tier = 0; tier < storage_tiers; tier++) {

425 - if (!multidb_ctx[tier] || finished_tiers[tier])

426 - continue;

427 -

428 - waiting_tiers++;

429 - if (completion_timedwait_for(&multidb_ctx[tier]->quiesce.completion, 1)) {

430 - completion_destroy(&multidb_ctx[tier]->quiesce.completion);

431 - finished_tiers[tier] = true;

432 - waiting_tiers--;

433 - nd_log(NDLS_DAEMON, NDLP_INFO, "DBENGINE tier %zu finished!", tier);

434 - }

435 - else if(iterations % 10 == 0) {

436 - pgc_main_stats = pgc_get_statistics(main_cache);

437 - nd_log(NDLS_DAEMON, NDLP_INFO,

438 - "Still waiting for DBENGINE tier %zu to finish "

439 - "(cache still has %zu pages, %zu bytes hot, for all tiers)...",

440 - tier,

441 - pgc_main_stats.queues[PGC_QUEUE_HOT].entries + pgc_main_stats.queues[PGC_QUEUE_DIRTY].entries,

442 - pgc_main_stats.queues[PGC_QUEUE_HOT].size + pgc_main_stats.queues[PGC_QUEUE_DIRTY].size);

443 - }

444 - }

445 - } while(waiting_tiers);

446 - nd_log(NDLS_DAEMON, NDLP_INFO, "DBENGINE shutdown completed...");

447 - }

448 -#endif

449 - watcher_step_complete(WATCHER_STEP_ID_FLUSH_DBENGINE_TIERS);

450 -

451 - rrd_finalize_collection_for_all_hosts();

452 - watcher_step_complete(WATCHER_STEP_ID_STOP_COLLECTION_FOR_ALL_HOSTS);

453 -

454 -#ifdef ENABLE_DBENGINE

455 - if(dbengine_enabled) {

456 - size_t running = 1;

457 - size_t count = 10;

458 - while(running && count) {

459 - running = 0;

460 - for (size_t tier = 0; tier < storage_tiers; tier++)

461 - running += rrdeng_collectors_running(multidb_ctx[tier]);

462 -

463 - if (running) {

464 - nd_log_limit_static_thread_var(erl, 1, 100 * USEC_PER_MS);

465 - nd_log_limit(&erl, NDLS_DAEMON, NDLP_NOTICE, "waiting for %zu collectors to finish", running);

466 - }

467 - count--;

468 - }

469 - watcher_step_complete(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_COLLECTORS_TO_FINISH);

470 -

471 - while (pgc_hot_and_dirty_entries(main_cache)) {

472 - pgc_flush_all_hot_and_dirty_pages(main_cache, PGC_SECTION_ALL);

473 - sleep_usec(100 * USEC_PER_MS);

474 - }

475 - watcher_step_complete(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_MAIN_CACHE_TO_FINISH_FLUSHING);

476 -

477 - for (size_t tier = 0; tier < storage_tiers; tier++)

478 - rrdeng_exit(multidb_ctx[tier]);

479 - rrdeng_enq_cmd(NULL, RRDENG_OPCODE_SHUTDOWN_EVLOOP, NULL, NULL, STORAGE_PRIORITY_BEST_EFFORT, NULL, NULL);

480 - watcher_step_complete(WATCHER_STEP_ID_STOP_DBENGINE_TIERS);

481 - }

482 - else {

483 - // Skip these steps

484 - watcher_step_complete(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_COLLECTORS_TO_FINISH);

485 - watcher_step_complete(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_MAIN_CACHE_TO_FINISH_FLUSHING);

486 - watcher_step_complete(WATCHER_STEP_ID_STOP_DBENGINE_TIERS);

487 - }

488 -#else

489 - // Skip these steps

490 - watcher_step_complete(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_COLLECTORS_TO_FINISH);

491 - watcher_step_complete(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_MAIN_CACHE_TO_FINISH_FLUSHING);

492 - watcher_step_complete(WATCHER_STEP_ID_STOP_DBENGINE_TIERS);

493 -#endif

494 -

495 - metadata_sync_shutdown();

496 - watcher_step_complete(WATCHER_STEP_ID_STOP_METASYNC_THREADS);

497 - }

498 -

499 - // Don't register a shutdown event if we crashed

500 - if (!ret)

501 - add_agent_event(EVENT_AGENT_SHUTDOWN_TIME, (int64_t)(now_monotonic_usec() - shutdown_start_time));

502 - sqlite_close_databases();

503 - watcher_step_complete(WATCHER_STEP_ID_CLOSE_SQL_DATABASES);

504 - sqlite_library_shutdown();

505 -

506 -

507 - // unlink the pid

508 - if(pidfile && *pidfile) {

509 - if(unlink(pidfile) != 0)

510 - netdata_log_error("EXIT: cannot unlink pidfile '%s'.", pidfile);

511 - }

512 - watcher_step_complete(WATCHER_STEP_ID_REMOVE_PID_FILE);

513 -

514 - netdata_ssl_cleanup();

515 - watcher_step_complete(WATCHER_STEP_ID_FREE_OPENSSL_STRUCTURES);

516 -

517 - (void) unlink(agent_incomplete_shutdown_file);

518 - watcher_step_complete(WATCHER_STEP_ID_REMOVE_INCOMPLETE_SHUTDOWN_FILE);

519 -

520 - watcher_shutdown_end();

521 - watcher_thread_stop();

522 - curl_global_cleanup();

523 -

524 -#ifdef OS_WINDOWS

525 - return;

526 -#endif

527 -

528 -#ifdef ENABLE_SENTRY

529 - nd_sentry_fini();

530 -#endif

531 -

532 - exit(ret);

533 -}

534 -

32 static void set_nofile_limit(struct rlimit *rl) {

33 // get the num files allowed

34 if(getrlimit(RLIMIT_NOFILE, rl) != 0) {

@@ -540,7 +37,7 @@ static void set_nofile_limit(struct rlimit *rl) {

37 }

39 netdata_log_info("resources control: allowed file descriptors: soft = %zu, max = %zu",

543 - (size_t) rl->rlim_cur, (size_t) rl->rlim_max);

40 + (size_t) rl->rlim_cur, (size_t) rl->rlim_max);

42 // make the soft/hard limits equal

43 rl->rlim_cur = rl->rlim_max;

@@ -558,57 +55,6 @@ static void set_nofile_limit(struct rlimit *rl) {

55 netdata_log_error("Number of open file descriptors allowed for this process is too low (RLIMIT_NOFILE=%zu)", (size_t)rl->rlim_cur);

56 }

561 -void cancel_main_threads() {

562 - nd_log_limits_unlimited();

563 -

564 - if (!static_threads)

565 - return;

566 -

567 - int i, found = 0;

568 - usec_t max = 5 * USEC_PER_SEC, step = 100000;

569 - for (i = 0; static_threads[i].name != NULL ; i++) {

570 - if (static_threads[i].enabled == NETDATA_MAIN_THREAD_RUNNING) {

571 - if (static_threads[i].thread) {

572 - netdata_log_info("EXIT: Stopping main thread: %s", static_threads[i].name);

573 - nd_thread_signal_cancel(static_threads[i].thread);

574 - } else {

575 - netdata_log_info("EXIT: No thread running (marking as EXITED): %s", static_threads[i].name);

576 - static_threads[i].enabled = NETDATA_MAIN_THREAD_EXITED;

577 - }

578 - found++;

579 - }

580 - }

581 -

582 - while(found && max > 0) {

583 - max -= step;

584 - netdata_log_info("Waiting %d threads to finish...", found);

585 - sleep_usec(step);

586 - found = 0;

587 - for (i = 0; static_threads[i].name != NULL ; i++) {

588 - if (static_threads[i].enabled == NETDATA_MAIN_THREAD_EXITED)

589 - continue;

590 -

591 - // Don't wait ourselves.

592 - if (nd_thread_is_me(static_threads[i].thread))

593 - continue;

594 -

595 - found++;

596 - }

597 - }

598 -

599 - if(found) {

600 - for (i = 0; static_threads[i].name != NULL ; i++) {

601 - if (static_threads[i].enabled != NETDATA_MAIN_THREAD_EXITED)

602 - netdata_log_error("Main thread %s takes too long to exit. Giving up...", static_threads[i].name);

603 - }

604 - }

605 - else

606 - netdata_log_info("All threads finished.");

607 -

608 - freez(static_threads);

609 - static_threads = NULL;

610 -}

611 -

58 static const struct option_def {

59 const char val;

60 const char *description;

src/daemon/main.h

+2 -34

@@ -4,39 +4,7 @@

4 #define NETDATA_MAIN_H 1

6 #include "common.h"

7 -

8 -extern struct config netdata_config;

9 -

10 -void cancel_main_threads(void);

11 -

12 -typedef enum {

13 - ABILITY_DATA_QUERIES = (1 << 0),

14 - ABILITY_WEB_REQUESTS = (1 << 1),

15 - ABILITY_STREAMING_CONNECTIONS = (1 << 2),

16 - SERVICE_MAINTENANCE = (1 << 3),

17 - SERVICE_COLLECTORS = (1 << 4),

18 - SERVICE_REPLICATION = (1 << 5),

19 - SERVICE_WEB_SERVER = (1 << 6),

20 - SERVICE_ACLK = (1 << 7),

21 - SERVICE_HEALTH = (1 << 8),

22 - SERVICE_STREAMING = (1 << 9),

23 - SERVICE_CONTEXT = (1 << 10),

24 - SERVICE_ANALYTICS = (1 << 11),

25 - SERVICE_EXPORTERS = (1 << 12),

26 - SERVICE_HTTPD = (1 << 13)

27 -} SERVICE_TYPE;

28 -

29 -typedef enum {

30 - SERVICE_THREAD_TYPE_NETDATA,

31 - SERVICE_THREAD_TYPE_LIBUV,

32 - SERVICE_THREAD_TYPE_EVENT_LOOP,

33 -} SERVICE_THREAD_TYPE;

34 -

35 -typedef void (*force_quit_t)(void *data);

36 -typedef void (*request_quit_t)(void *data);

37 -

38 -void service_exits(void);

39 -bool service_running(SERVICE_TYPE service);

40 -struct service_thread *service_register(SERVICE_THREAD_TYPE thread_type, request_quit_t request_quit_callback, force_quit_t force_quit_callback, void *data, bool update __maybe_unused);

7 +#include "daemon-service.h"

8 +#include "daemon-shutdown.h"

10 #endif /* NETDATA_MAIN_H */

src/daemon/pulse/pulse-aral.c

+9 -5

@@ -19,7 +19,7 @@ static struct {

19 ARAL_STATS_JudyLSet idx;

20 } globals = { 0 };

22 -static void pulse_aral_register_statistics(struct aral_statistics *stats, const char *name) {

22 +void pulse_aral_register_statistics(struct aral_statistics *stats, const char *name) {

23 if(!name || !stats)

24 return;

@@ -60,6 +60,7 @@ void pulse_aral_unregister(ARAL *ar) {

61 void pulse_aral_init(void) {

62 pulse_aral_register_statistics(aral_by_size_statistics(), "by-size");

63 + pulse_aral_register_statistics(judy_aral_statistics(), "judy");

64 }

66 void pulse_aral_do(bool extended) {

@@ -86,14 +87,17 @@ void pulse_aral_do(bool extended) {

87 mmap_allocated_bytes = mmap_used_bytes;

88 size_t mmap_free_bytes = mmap_allocated_bytes - mmap_used_bytes;

90 + size_t allocated_total = malloc_allocated_bytes + mmap_allocated_bytes;

91 + size_t used_total = malloc_used_bytes + mmap_used_bytes;

92 +

93 size_t structures_bytes = __atomic_load_n(&stats->structures.allocated_bytes, __ATOMIC_RELAXED);

95 size_t padding_bytes = __atomic_load_n(&stats->malloc.padding_bytes, __ATOMIC_RELAXED) +

96 __atomic_load_n(&stats->mmap.padding_bytes, __ATOMIC_RELAXED);

98 NETDATA_DOUBLE utilization;

95 - if((malloc_used_bytes + mmap_used_bytes != 0) && (malloc_allocated_bytes + mmap_allocated_bytes != 0))

96 - utilization = 100.0 * (NETDATA_DOUBLE)(malloc_used_bytes + mmap_used_bytes) / (NETDATA_DOUBLE)(malloc_allocated_bytes + mmap_allocated_bytes);

99 + if(allocated_total)

100 + utilization = 100.0 * (NETDATA_DOUBLE)used_total / (NETDATA_DOUBLE)allocated_total;

101 else

102 utilization = 100.0;

103

@@ -160,10 +164,10 @@ void pulse_aral_do(bool extended) {

164

165 rrdlabels_add(ai->st_utilization->rrdlabels, "ARAL", ai->name, RRDLABEL_SRC_AUTO);

166

163 - ai->rd_utilization = rrddim_add(ai->st_utilization, "utilization", NULL, 1, 10000, RRD_ALGORITHM_ABSOLUTE);

167 + ai->rd_utilization = rrddim_add(ai->st_utilization, "utilization", NULL, 1, 1000, RRD_ALGORITHM_ABSOLUTE);

168 }

169

166 - rrddim_set_by_pointer(ai->st_utilization, ai->rd_utilization, (collected_number)(utilization * 10000.0));

170 + rrddim_set_by_pointer(ai->st_utilization, ai->rd_utilization, (collected_number)utilization * 1000LL);

171 rrdset_done(ai->st_utilization);

172 }

173 }

src/daemon/pulse/pulse-aral.h

@@ -5,6 +5,7 @@

6 #include "daemon/common.h"

8 +void pulse_aral_register_statistics(struct aral_statistics *stats, const char *name);

9 void pulse_aral_register(ARAL *ar, const char *name);

10 void pulse_aral_unregister(ARAL *ar);

src/daemon/pulse/pulse-workers.c

+1 -1

@@ -113,7 +113,7 @@ struct worker_utilization {

113

114 static struct worker_utilization all_workers_utilization[] = {

115 { .name = "PULSE", .family = "workers pulse", .priority = 1000000 },

116 - { .name = "HEALTH", .family = "workers health alarms", .priority = 1000000 },

116 + { .name = "HEALTH", .family = "workers health alerts", .priority = 1000000 },

117 { .name = "MLTRAIN", .family = "workers ML training", .priority = 1000000 },

118 { .name = "MLDETECT", .family = "workers ML detection", .priority = 1000000 },

119 { .name = "STREAM", .family = "workers streaming", .priority = 1000000 },

src/daemon/pulse/pulse.c

-3

@@ -55,7 +55,6 @@ static void pulse_cleanup(void *pptr)

56 pulse_workers_cleanup();

57 worker_unregister();

58 - netdata_log_info("cleaning up...");

59 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

60 }

@@ -158,8 +157,6 @@ static void pulse_thread_sqlite3_cleanup(void *pptr)

157

158 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

159

161 - netdata_log_info("cleaning up...");

162 -

160 worker_unregister();

161

162 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

src/daemon/service.c

-1

@@ -268,7 +268,6 @@ static void service_main_cleanup(void *pptr)

268

269 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

270

271 - netdata_log_debug(D_SYSTEM, "Cleaning up...");

271 worker_unregister();

272

273 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

src/database/engine/cache.c

+20 -16

@@ -580,8 +580,12 @@ static void pgc_section_pages_static_aral_init(void) {

580

581 spinlock_lock(&spinlock);

582

583 - if(!pgc_sections_aral)

584 - pgc_sections_aral = aral_by_size_acquire(sizeof(struct section_pages));

583 + if(!pgc_sections_aral) {

584 + pgc_sections_aral = aral_create(

585 + "pgc-sections", sizeof(struct section_pages), 0, 0, &pgc_aral_statistics, NULL, NULL, false, false);

586 +

587 + pulse_aral_register_statistics(&pgc_aral_statistics, "pgc");

588 + }

589

590 spinlock_unlock(&spinlock);

591 }

@@ -1182,16 +1186,18 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic

1186 else if(unlikely(wait)) {

1187 // evict as many as necessary for the cache to go at the predefined threshold

1188 per1000 = cache_usage_per1000(cache, &max_size_to_evict);

1185 - max_size_to_evict /= 2; // do it in 2 steps

1189 + max_size_to_evict /= 3; // do it in 3 steps

1190 if(per1000 >= cache->config.severe_pressure_per1000) {

1191 under_sever_pressure = true;

1188 - max_pages_to_evict = max_pages_to_evict ? max_pages_to_evict * 2 : 4096;

1189 - // max_pages_to_evict = 1;

1192 + max_pages_to_evict = max_pages_to_evict ? max_pages_to_evict * 2 : 512;

1193 + if(max_pages_to_evict > 4096)

1194 + max_pages_to_evict = 4096;

1195 }

1196 else if(per1000 >= cache->config.aggressive_evict_per1000) {

1197 under_sever_pressure = false;

1193 - max_pages_to_evict = max_pages_to_evict ? max_pages_to_evict * 2 : 128;

1194 - // max_pages_to_evict = 1;

1198 + max_pages_to_evict = max_pages_to_evict ? max_pages_to_evict * 2 : 32;

1199 + if(max_pages_to_evict > 1024)

1200 + max_pages_to_evict = 1024;

1201 }

1202 else {

1203 under_sever_pressure = false;

@@ -1948,7 +1954,8 @@ static void *pgc_evict_thread(void *ptr) {

1954 worker_register_job_name(0, "signaled");

1955 worker_register_job_name(1, "scheduled");

1956

1951 - unsigned job_id = 0, severe_pressure_counter = 0;

1957 + unsigned job_id = 0;

1958 + usec_t last_malloc_release_ut = 0;

1959

1960 while (true) {

1961 worker_is_idle();

@@ -1965,18 +1972,17 @@ static void *pgc_evict_thread(void *ptr) {

1972

1973 size_t size_to_evict = 0;

1974 if(cache_usage_per1000(cache, &size_to_evict) > cache->config.severe_pressure_per1000) {

1968 - severe_pressure_counter++;

1975 + usec_t now_ut = now_monotonic_usec();

1976 +

1977 + if(last_malloc_release_ut + USEC_PER_SEC < now_ut) {

1978 + last_malloc_release_ut = now_ut;

1979

1970 - if(severe_pressure_counter > 100) {

1971 - // so, we tried 100 times to reduce memory,

1980 + // so, we tried 100 times to reduce memory, and a second has passed,

1981 // but it is still severe!

1982

1983 mallocz_release_as_much_memory_to_the_system();

1975 - severe_pressure_counter = 0;

1984 }

1985 }

1978 - else

1979 - severe_pressure_counter = 0;

1986 }

1987

1988 worker_unregister();

@@ -2066,8 +2072,6 @@ PGC *pgc_create(const char *name,

2072 false);

2073 }

2074 }

2069 -

2070 - pulse_aral_register(cache->index[0].aral, "pgc");

2075 #endif

2076

2077

src/database/engine/dbengine-stresstest.c

+1 -1

@@ -447,7 +447,7 @@ void dbengine_stress_test(unsigned TEST_DURATION_SEC, unsigned DSET_CHARTS, unsi

447 }

448 freez(query_threads);

449 rrd_wrlock();

450 - rrdeng_prepare_exit((struct rrdengine_instance *)host->db[0].si);

450 + rrdeng_quiesce((struct rrdengine_instance *)host->db[0].si);

451 rrdeng_exit((struct rrdengine_instance *)host->db[0].si);

452 rrdeng_enq_cmd(NULL, RRDENG_OPCODE_SHUTDOWN_EVLOOP, NULL, NULL, STORAGE_PRIORITY_BEST_EFFORT, NULL, NULL);

453 rrd_wrunlock();

src/database/engine/dbengine-unittest.c

+1 -1

@@ -408,7 +408,7 @@ int test_dbengine(void) {

408 }

409

410 rrd_wrlock();

411 - rrdeng_prepare_exit((struct rrdengine_instance *)host->db[0].si);

411 + rrdeng_quiesce((struct rrdengine_instance *)host->db[0].si);

412 rrdeng_exit((struct rrdengine_instance *)host->db[0].si);

413 rrdeng_enq_cmd(NULL, RRDENG_OPCODE_SHUTDOWN_EVLOOP, NULL, NULL, STORAGE_PRIORITY_BEST_EFFORT, NULL, NULL);

414 rrd_wrunlock();

src/database/engine/metric.c

+1 -1

@@ -380,7 +380,7 @@ inline MRG *mrg_create(ssize_t partitions) {

380

381 mrg->index[i].aral = aral_create(buf, sizeof(METRIC), 0, 16384, &mrg_aral_statistics, NULL, NULL, false, false);

382 }

383 - pulse_aral_register(mrg->index[0].aral, "mrg");

383 + pulse_aral_register_statistics(&mrg_aral_statistics, "mrg");

384

385 return mrg;

386 }

src/database/engine/page.c

+1 -1

@@ -200,7 +200,7 @@ void pgd_init_arals(void) {

200 pgd_alloc_globals.sizeof_gorilla_writer_t = aral_actual_element_size(pgd_alloc_globals.aral_gorilla_writer[0]);

201 pgd_alloc_globals.sizeof_gorilla_buffer_32bit = aral_actual_element_size(pgd_alloc_globals.aral_gorilla_buffer[0]);

202

203 - pulse_aral_register(pgd_alloc_globals.aral_pgd[0], "pgd");

203 + pulse_aral_register_statistics(&pgd_aral_statistics, "pgd");

204 }

205

206 static ARAL *pgd_get_aral_by_size_and_partition(size_t size, size_t partition) {

src/database/engine/pagecache.c

+3 -1

@@ -783,6 +783,8 @@ void rrdeng_prep_query(struct page_details_control *pdc, bool worker) {

783 usec_t start_ut = now_monotonic_usec();

784 if(likely(pdc->priority == STORAGE_PRIORITY_SYNCHRONOUS))

785 pdc_route_synchronously(pdc->ctx, pdc);

786 + else if(likely(pdc->priority == STORAGE_PRIORITY_SYNCHRONOUS_FIRST))

787 + pdc_route_synchronously_first(pdc->ctx, pdc);

788 else

789 pdc_route_asynchronously(pdc->ctx, pdc);

790 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.prep_time_to_route, now_monotonic_usec() - start_ut, __ATOMIC_RELAXED);

@@ -827,7 +829,7 @@ void pg_cache_preload(struct rrdeng_query_handle *handle) {

829 if(ctx_is_available_for_queries(handle->ctx)) {

830 handle->pdc->refcount++; // we get 1 for the query thread and 1 for the prep thread

831

830 - if(unlikely(handle->pdc->priority == STORAGE_PRIORITY_SYNCHRONOUS))

832 + if(unlikely(handle->pdc->priority == STORAGE_PRIORITY_SYNCHRONOUS || handle->pdc->priority == STORAGE_PRIORITY_SYNCHRONOUS_FIRST))

833 rrdeng_prep_query(handle->pdc, false);

834 else

835 rrdeng_enq_cmd(handle->ctx, RRDENG_OPCODE_QUERY, handle->pdc, NULL, handle->priority, NULL, NULL);

src/database/engine/rrdengine.c

+9 -2

@@ -265,7 +265,7 @@ void page_descriptors_init(void) {

265 NULL,

266 NULL, NULL, false, false);

267

268 - pulse_aral_register(rrdeng_main.xt_io_descr.ar, "descriptors");

268 + pulse_aral_register(rrdeng_main.descriptors.ar, "descriptors");

269 }

270

271 struct page_descr_with_data *page_descriptor_get(void) {

@@ -1314,7 +1314,10 @@ static void after_flush_all_hot_and_dirty_pages_of_section(struct rrdengine_inst

1314 static void *flush_all_hot_and_dirty_pages_of_section_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {

1315 worker_is_busy(UV_EVENT_DBENGINE_QUIESCE);

1316 pgc_flush_all_hot_and_dirty_pages(main_cache, (Word_t)ctx);

1317 - completion_mark_complete(&ctx->quiesce.completion);

1317 +

1318 + for(size_t i = 0; i < pgc_max_flushers() ; i++)

1319 + rrdeng_enq_cmd(NULL, RRDENG_OPCODE_FLUSH_MAIN, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

1320 +

1321 return data;

1322 }

1323

@@ -1499,6 +1502,10 @@ void pdc_route_synchronously(struct rrdengine_instance *ctx, struct page_details

1502 pdc_to_epdl_router(ctx, pdc, epdl_populate_pages_synchronously, epdl_populate_pages_synchronously);

1503 }

1504

1505 +void pdc_route_synchronously_first(struct rrdengine_instance *ctx, struct page_details_control *pdc) {

1506 + pdc_to_epdl_router(ctx, pdc, epdl_populate_pages_synchronously, epdl_populate_pages_asynchronously);

1507 +}

1508 +

1509 #define MAX_RETRIES_TO_START_INDEX (100)

1510 static void *journal_v2_indexing_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {

1511 unsigned count = 0;

src/database/engine/rrdengine.h

+1 -1

@@ -448,7 +448,6 @@ struct rrdengine_instance {

448 struct {

449 bool exit_mode;

450 bool enabled; // when set (before shutdown), queries are prohibited

451 - struct completion completion;

451 } quiesce;

452

453 struct {

@@ -527,6 +526,7 @@ void rrdeng_enq_cmd(struct rrdengine_instance *ctx, enum rrdeng_opcode opcode, v

526

527 void pdc_route_asynchronously(struct rrdengine_instance *ctx, struct page_details_control *pdc);

528 void pdc_route_synchronously(struct rrdengine_instance *ctx, struct page_details_control *pdc);

529 +void pdc_route_synchronously_first(struct rrdengine_instance *ctx, struct page_details_control *pdc);

530

531 void pdc_acquire(PDC *pdc);

532 bool pdc_release_and_destroy_if_unreferenced(PDC *pdc, bool worker, bool router);

src/database/engine/rrdengineapi.c

+2 -7

@@ -1125,9 +1125,6 @@ void rrdeng_readiness_wait(struct rrdengine_instance *ctx) {

1125 netdata_log_info("DBENGINE: tier %d is ready for data collection and queries", ctx->config.tier);

1126 }

1127

1128 -void rrdeng_exit_mode(struct rrdengine_instance *ctx) {

1129 - __atomic_store_n(&ctx->quiesce.exit_mode, true, __ATOMIC_RELAXED);

1130 -}

1128 /*

1129 * Returns 0 on success, negative on error

1130 */

@@ -1228,13 +1225,12 @@ int rrdeng_exit(struct rrdengine_instance *ctx) {

1225 count--;

1226 }

1227

1231 - netdata_log_info("DBENGINE: flushing main cache for tier %d", ctx->config.tier);

1228 pgc_flush_all_hot_and_dirty_pages(main_cache, (Word_t)ctx);

1229

1234 - netdata_log_info("DBENGINE: shutting down tier %d", ctx->config.tier);

1230 struct completion completion = {};

1231 completion_init(&completion);

1232 rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_SHUTDOWN, NULL, &completion, STORAGE_PRIORITY_BEST_EFFORT, NULL, NULL);

1233 +

1234 completion_wait_for(&completion);

1235 completion_destroy(&completion);

1236

@@ -1247,14 +1243,13 @@ int rrdeng_exit(struct rrdengine_instance *ctx) {

1243 return 0;

1244 }

1245

1250 -void rrdeng_prepare_exit(struct rrdengine_instance *ctx) {

1246 +void rrdeng_quiesce(struct rrdengine_instance *ctx) {

1247 if (NULL == ctx)

1248 return;

1249

1250 // FIXME - ktsaou - properly cleanup ctx

1251 // 1. make sure all collectors are stopped

1252

1257 - completion_init(&ctx->quiesce.completion);

1253 rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_QUIESCE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

1254 }

1255

src/database/engine/rrdengineapi.h

+1 -2

@@ -71,10 +71,9 @@ int rrdeng_init(

71 time_t max_retention_s);

73 void rrdeng_readiness_wait(struct rrdengine_instance *ctx);

74 -void rrdeng_exit_mode(struct rrdengine_instance *ctx);

75 int rrdeng_exit(struct rrdengine_instance *ctx);

77 -void rrdeng_prepare_exit(struct rrdengine_instance *ctx);

76 +void rrdeng_quiesce(struct rrdengine_instance *ctx);

77 bool rrdeng_metric_retention_by_uuid(STORAGE_INSTANCE *si, nd_uuid_t *dim_uuid, time_t *first_entry_s, time_t *last_entry_s);

79 extern STORAGE_METRICS_GROUP *rrdeng_metrics_group_get(STORAGE_INSTANCE *si, nd_uuid_t *uuid);

src/database/rrd.h

@@ -61,6 +61,7 @@ typedef enum __attribute__ ((__packed__)) storage_priority {

62 // synchronous query, not to be dispatched to workers or queued

63 STORAGE_PRIORITY_SYNCHRONOUS,

64 + STORAGE_PRIORITY_SYNCHRONOUS_FIRST,

66 STORAGE_PRIORITY_INTERNAL_MAX_DONT_USE,

67 } STORAGE_PRIORITY;

src/database/sqlite/sqlite_aclk_node.c

+33 -9

@@ -122,6 +122,10 @@ void aclk_check_node_info_and_collectors(void)

122 size_t replicating = 0;

123 size_t context_pp = 0;

124

125 + STRING *context_loading_host = NULL;

126 + STRING *replicating_host = NULL;

127 + STRING *context_pp_host = NULL;

128 +

129 time_t now = now_realtime_sec();

130 dfe_start_reentrant(rrdhost_root_index, host)

131 {

@@ -132,6 +136,7 @@ void aclk_check_node_info_and_collectors(void)

136 if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))) {

137 internal_error(true, "ACLK SYNC: Context still pending for %s", rrdhost_hostname(host));

138 context_loading++;

139 + context_loading_host = host->hostname;

140 continue;

141 }

142

@@ -141,13 +146,16 @@ void aclk_check_node_info_and_collectors(void)

146 if (unlikely(rrdhost_receiver_replicating_charts(host))) {

147 internal_error(true, "ACLK SYNC: Host %s is still replicating", rrdhost_hostname(host));

148 replicating++;

149 + replicating_host = host->hostname;

150 continue;

151 }

152

153 bool pp_queue_empty = !(host->rrdctx.pp_queue && dictionary_entries(host->rrdctx.pp_queue));

154

149 - if (!pp_queue_empty && (wc->node_info_send_time || wc->node_collectors_send))

155 + if (!pp_queue_empty && (wc->node_info_send_time || wc->node_collectors_send)) {

156 context_pp++;

157 + context_pp_host = host->hostname;

158 + }

159

160 if (pp_queue_empty && wc->node_info_send_time && wc->node_info_send_time + 30 < now) {

161 wc->node_info_send_time = 0;

@@ -165,14 +173,30 @@ void aclk_check_node_info_and_collectors(void)

173 dfe_done(host);

174

175 if (context_loading || replicating || context_pp) {

176 + const char *context_loading_pre = "", *context_loading_body = "", *context_loading_post = "";

177 + if(context_loading == 1) {

178 + context_loading_pre = " (host '";

179 + context_loading_body = string2str(context_loading_host);

180 + context_loading_post = "')";

181 + }

182 + const char *replicating_pre = "", *replicating_body = "", *replicating_post = "";

183 + if(replicating == 1) {

184 + replicating_pre = " (host '";

185 + replicating_body = string2str(replicating_host);

186 + replicating_post = "')";

187 + }

188 + const char *context_pp_pre = "", *context_pp_body = "", *context_pp_post = "";

189 + if(context_pp == 1) {

190 + context_pp_pre = " (host '";

191 + context_pp_body = string2str(context_pp_host);

192 + context_pp_post = "')";

193 + }

194 nd_log_limit_static_thread_var(erl, 10, 100 * USEC_PER_MS);

169 - nd_log_limit(

170 - &erl,

171 - NDLS_DAEMON,

172 - NDLP_INFO,

173 - "%zu nodes loading contexts, %zu replicating data, %zu pending context post processing",

174 - context_loading,

175 - replicating,

176 - context_pp);

195 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_INFO,

196 + "%zu nodes loading contexts%s%s%s, %zu replicating data%s%s%s, %zu pending context post processing%s%s%s",

197 + context_loading, context_loading_pre, context_loading_body, context_loading_post,

198 + replicating, replicating_pre, replicating_body, replicating_post,

199 + context_pp, context_pp_pre, context_pp_body, context_pp_post

200 + );

201 }

202 }

src/database/sqlite/sqlite_metadata.c

+28 -5

@@ -225,6 +225,7 @@ struct metadata_wc {

225 /* FIFO command queue */

226 SPINLOCK cmd_queue_lock;

227 struct metadata_cmd *cmd_base;

228 + ARAL *ar;

229 };

230

231 #define metadata_flag_check(target_flags, flag) (__atomic_load_n(&((target_flags)->flags), __ATOMIC_SEQ_CST) & (flag))

@@ -447,7 +448,8 @@ struct node_instance_list *get_node_list(void)

448 continue;

449

450 if (rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD)) {

450 - netdata_log_info(

451 + nd_log_limit_static_global_var(erl, 1, 0);

452 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_INFO,

453 "ACLK: 'host:%s' skipping get node list because context is initializing", rrdhost_hostname(host));

454 continue;

455 }

@@ -1490,7 +1492,7 @@ static void metadata_free_cmd_queue(struct metadata_wc *wc)

1492 while(wc->cmd_base) {

1493 struct metadata_cmd *t = wc->cmd_base;

1494 DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(wc->cmd_base, t, prev, next);

1493 - freez(t);

1495 + aral_freez(wc->ar, t);

1496 }

1497 spinlock_unlock(&wc->cmd_queue_lock);

1498 }

@@ -1505,7 +1507,7 @@ static void metadata_enq_cmd(struct metadata_wc *wc, struct metadata_cmd *cmd)

1507 if (unlikely(metadata_flag_check(wc, METADATA_FLAG_SHUTDOWN)))

1508 goto wakeup_event_loop;

1509

1508 - struct metadata_cmd *t = mallocz(sizeof(*t));

1510 + struct metadata_cmd *t = aral_mallocz(wc->ar);

1511 *t = *cmd;

1512 t->prev = t->next = NULL;

1513

@@ -1519,14 +1521,14 @@ wakeup_event_loop:

1521

1522 static struct metadata_cmd metadata_deq_cmd(struct metadata_wc *wc)

1523 {

1522 - struct metadata_cmd ret;

1524 + struct metadata_cmd ret, *to_free = NULL;

1525

1526 spinlock_lock(&wc->cmd_queue_lock);

1527 if(wc->cmd_base) {

1528 struct metadata_cmd *t = wc->cmd_base;

1529 DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(wc->cmd_base, t, prev, next);

1530 ret = *t;

1529 - freez(t);

1531 + to_free = t;

1532 }

1533 else {

1534 ret.opcode = METADATA_DATABASE_NOOP;

@@ -1534,6 +1536,8 @@ static struct metadata_cmd metadata_deq_cmd(struct metadata_wc *wc)

1536 }

1537 spinlock_unlock(&wc->cmd_queue_lock);

1538

1539 + aral_freez(wc->ar, to_free);

1540 +

1541 return ret;

1542 }

1543

@@ -2063,6 +2067,8 @@ static void metadata_event_loop(void *arg)

2067 struct metadata_wc *wc = arg;

2068 enum metadata_opcode opcode;

2069

2070 + wc->ar = aral_by_size_acquire(sizeof(struct metadata_cmd));

2071 +

2072 uv_thread_set_name_np("METASYNC");

2073 loop = wc->loop = mallocz(sizeof(uv_loop_t));

2074 ret = uv_loop_init(loop);

@@ -2263,6 +2269,7 @@ error_after_async_init:

2269 fatal_assert(0 == uv_loop_close(loop));

2270 error_after_loop_init:

2271 freez(loop);

2272 + aral_by_size_release(wc->ar);

2273 worker_unregister();

2274 }

2275

@@ -2316,6 +2323,22 @@ void metadata_sync_shutdown_prepare(void)

2323 nd_log(NDLS_DAEMON, NDLP_DEBUG, "METADATA: Host scan complete; can continue with shutdown");

2324 }

2325

2326 +void *metadata_sync_shutdown_thread(void *ptr __maybe_unused) {

2327 + metadata_sync_shutdown_prepare();

2328 + return NULL;

2329 +}

2330 +

2331 +static ND_THREAD *metdata_sync_shutdown_background_wait_thread = NULL;

2332 +void metadata_sync_shutdown_background(void) {

2333 + metdata_sync_shutdown_background_wait_thread = nd_thread_create(

2334 + "METASYNC-SHUTDOWN", NETDATA_THREAD_OPTION_JOINABLE, metadata_sync_shutdown_thread, NULL);

2335 +}

2336 +

2337 +void metadata_sync_shutdown_background_wait(void) {

2338 + nd_thread_join(metdata_sync_shutdown_background_wait_thread);

2339 + metadata_sync_shutdown();

2340 +}

2341 +

2342 // -------------------------------------------------------------

2343 // Init function called on agent startup

2344

src/database/sqlite/sqlite_metadata.h

@@ -70,6 +70,9 @@ void metadata_queue_ae_save(RRDHOST *host, ALARM_ENTRY *ae);

70 void metadata_queue_ae_deletion(ALARM_ENTRY *ae);

71 void commit_alert_transitions(RRDHOST *host);

73 +void metadata_sync_shutdown_background(void);

74 +void metadata_sync_shutdown_background_wait(void);

75 +

76 // UNIT TEST

77 int metadata_unittest(void);

78 #endif //NETDATA_SQLITE_METADATA_H

src/exporting/exporting_engine.c

-2

@@ -124,8 +124,6 @@ static void exporting_main_cleanup(void *pptr)

124

125 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

126

127 - netdata_log_info("cleaning up...");

128 -

127 if (!engine) {

128 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

129 return;

src/health/health_event_loop.c

+16 -25

@@ -660,39 +660,31 @@ static void health_event_loop(void) {

660 health_alarm_log_process_to_send_notifications(host, hrm);

661 alerts_raised_summary_free(hrm);

662

663 - if (unlikely(!service_running(SERVICE_HEALTH))) {

664 - // wait for all notifications to finish before allowing health to be cleaned up

665 - wait_for_all_notifications_to_finish_before_allowing_health_to_be_cleaned_up();

666 - break;

667 - }

668 - }

663 + int32_t pending = __atomic_load_n(&host->health.pending_transitions, __ATOMIC_RELAXED);

664 + if (pending)

665 + commit_alert_transitions(host);

666

670 - int32_t pending = __atomic_load_n(&host->health.pending_transitions, __ATOMIC_RELAXED);

671 - if (pending)

672 - commit_alert_transitions(host);

673 -

674 - if (!__atomic_load_n(&host->health.pending_transitions, __ATOMIC_RELAXED)) {

675 - struct aclk_sync_cfg_t *wc = host->aclk_config;

676 - if (wc && wc->send_snapshot == 1) {

677 - wc->send_snapshot = 2;

678 - rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);

679 - } else {

680 - if (process_alert_pending_queue(host))

667 + if (!__atomic_load_n(&host->health.pending_transitions, __ATOMIC_RELAXED)) {

668 + struct aclk_sync_cfg_t *wc = host->aclk_config;

669 + if (wc && wc->send_snapshot == 1) {

670 + wc->send_snapshot = 2;

671 rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);

672 + } else {

673 + if (process_alert_pending_queue(host))

674 + rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);

675 + }

676 }

677 }

684 -

678 dfe_done(host);

679

687 - // wait for all notifications to finish before allowing health to be cleaned up

688 - wait_for_all_notifications_to_finish_before_allowing_health_to_be_cleaned_up();

689 -

680 if(unlikely(!service_running(SERVICE_HEALTH)))

681 break;

682

683 health_sleep(next_run, loop);

694 -

684 } // forever

685 +

686 + // wait for all notifications to finish before allowing health to be cleaned up

687 + wait_for_all_notifications_to_finish_before_allowing_health_to_be_cleaned_up();

688 }

689

690

@@ -702,7 +694,6 @@ static void health_main_cleanup(void *pptr) {

694

695 worker_unregister();

696 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

705 - netdata_log_info("cleaning up...");

697 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

698

699 nd_log(NDLS_DAEMON, NDLP_DEBUG, "Health thread ended.");

@@ -716,8 +707,8 @@ void *health_main(void *ptr) {

707 worker_register_job_name(WORKER_HEALTH_JOB_CALC_EVAL, "calc eval");

708 worker_register_job_name(WORKER_HEALTH_JOB_WARNING_EVAL, "warning eval");

709 worker_register_job_name(WORKER_HEALTH_JOB_CRITICAL_EVAL, "critical eval");

719 - worker_register_job_name(WORKER_HEALTH_JOB_ALARM_LOG_ENTRY, "alarm log entry");

720 - worker_register_job_name(WORKER_HEALTH_JOB_ALARM_LOG_PROCESS, "alarm log process");

710 + worker_register_job_name(WORKER_HEALTH_JOB_ALARM_LOG_ENTRY, "alert log entry");

711 + worker_register_job_name(WORKER_HEALTH_JOB_ALARM_LOG_PROCESS, "alert log process");

712 worker_register_job_name(WORKER_HEALTH_JOB_DELAYED_INIT_RRDSET, "rrdset init");

713 worker_register_job_name(WORKER_HEALTH_JOB_DELAYED_INIT_RRDDIM, "rrddim init");

714

src/libnetdata/functions_evloop/functions_evloop.c

+122 -116

@@ -59,6 +59,20 @@ struct functions_evloop_globals {

59 } dyncfg;

61 struct rrd_functions_expectation *expectations;

62 +

63 + struct buffered_reader reader;

64 + BUFFER *buffer;

65 + char *words[MAX_FUNCTION_PARAMETERS];

66 + struct {

67 + size_t last_len; // to remember the last pos - do not use a pointer, the buffer may realloc...

68 + bool enabled;

69 + char *transaction;

70 + char *function;

71 + char *timeout_s;

72 + char *access;

73 + char *source;

74 + char *content_type;

75 + } deferred;

76 };

78 static void rrd_functions_worker_canceller(void *data) {

@@ -137,7 +151,8 @@ static void worker_add_job(struct functions_evloop_globals *wg, const char *keyw

151 function?function:"(unset)");

152 }

153 else {

140 - // nd_log(NDLS_COLLECTORS, NDLP_INFO, "WORKER JOB WITH PAYLOAD '%s'", payload ? buffer_tostring(payload) : "NONE");

154 +// nd_log(NDLS_COLLECTORS, NDLP_INFO, "WORKER JOB: keyword '%s', transaction '%s', function '%s', timeout '%s', access '%s', source '%s', payload '%s'",

155 +// keyword, transaction, function, timeout_s, access, source, payload ? buffer_tostring(payload) : "NONE");

156

157 int timeout = str2i(timeout_s);

158

@@ -187,28 +202,116 @@ static void worker_add_job(struct functions_evloop_globals *wg, const char *keyw

202 }

203 }

204

205 +static bool rrd_function_worker_global_process_input(struct functions_evloop_globals *wg) {

206 + if(wg->deferred.enabled) {

207 + char *s = (char *)buffer_tostring(wg->buffer);

208 +

209 + if(strstr(&s[wg->deferred.last_len], PLUGINSD_CALL_FUNCTION_PAYLOAD_END "\n") != NULL) {

210 + // nd_log(NDLS_COLLECTORS, NDLP_INFO, "FUNCTION PAYLOAD END");

211 +

212 + if(wg->deferred.last_len > 0)

213 + // remove the trailing newline from the buffer

214 + wg->deferred.last_len--;

215 +

216 + s[wg->deferred.last_len] = '\0';

217 + wg->buffer->len = wg->deferred.last_len;

218 + wg->buffer->content_type = content_type_string2id(wg->deferred.content_type);

219 + worker_add_job(wg, PLUGINSD_CALL_FUNCTION_PAYLOAD_BEGIN,

220 + wg->deferred.transaction, wg->deferred.function,

221 + wg->deferred.timeout_s, wg->buffer, wg->deferred.access, wg->deferred.source);

222 + buffer_flush(wg->buffer);

223 +

224 + freez(wg->deferred.transaction);

225 + freez(wg->deferred.function);

226 + freez(wg->deferred.timeout_s);

227 + freez(wg->deferred.access);

228 + freez(wg->deferred.source);

229 + freez(wg->deferred.content_type);

230 + memset(&wg->deferred, 0, sizeof(wg->deferred));

231 + }

232 + else

233 + wg->deferred.last_len = wg->buffer->len;

234 +

235 + return false;

236 + }

237 +

238 + size_t num_words = quoted_strings_splitter_whitespace((char *)buffer_tostring(wg->buffer), wg->words, _countof(wg->words));

239 + const char *keyword = get_word(wg->words, num_words, 0);

240 +

241 + char **words = wg->words;

242 + if(keyword && (strcmp(keyword, PLUGINSD_CALL_FUNCTION) == 0)) {

243 + char *transaction = get_word(words, num_words, 1);

244 + char *timeout_s = get_word(words, num_words, 2);

245 + char *function = get_word(words, num_words, 3);

246 + char *access = get_word(words, num_words, 4);

247 + char *source = get_word(words, num_words, 5);

248 + worker_add_job(wg, keyword, transaction, function, timeout_s, NULL, access, source);

249 + }

250 + else if(keyword && (strcmp(keyword, PLUGINSD_CALL_FUNCTION_PAYLOAD_BEGIN) == 0)) {

251 + char *transaction = get_word(words, num_words, 1);

252 + char *timeout_s = get_word(words, num_words, 2);

253 + char *function = get_word(words, num_words, 3);

254 + char *access = get_word(words, num_words, 4);

255 + char *source = get_word(words, num_words, 5);

256 + char *content_type = get_word(words, num_words, 6);

257 +

258 + wg->deferred.transaction = strdupz(transaction ? transaction : "");

259 + wg->deferred.timeout_s = strdupz(timeout_s ? timeout_s : "");

260 + wg->deferred.function = strdupz(function ? function : "");

261 + wg->deferred.access = strdupz(access ? access : "");

262 + wg->deferred.source = strdupz(source ? source : "");

263 + wg->deferred.content_type = strdupz(content_type ? content_type : "");

264 + wg->deferred.last_len = 0;

265 + wg->deferred.enabled = true;

266 + }

267 + else if(keyword && strcmp(keyword, PLUGINSD_CALL_FUNCTION_CANCEL) == 0) {

268 + char *transaction = get_word(words, num_words, 1);

269 + const DICTIONARY_ITEM *acquired = dictionary_get_and_acquire_item(wg->worker_queue, transaction);

270 + if(acquired) {

271 + struct functions_evloop_worker_job *j = dictionary_acquired_item_value(acquired);

272 + __atomic_store_n(&j->cancelled, true, __ATOMIC_RELAXED);

273 + dictionary_acquired_item_release(wg->worker_queue, acquired);

274 + dictionary_del(wg->worker_queue, transaction);

275 + dictionary_garbage_collect(wg->worker_queue);

276 + }

277 + else

278 + nd_log(NDLS_COLLECTORS, NDLP_NOTICE, "Received CANCEL for transaction '%s', but it not available here", transaction);

279 + }

280 + else if(keyword && strcmp(keyword, PLUGINSD_CALL_FUNCTION_PROGRESS) == 0) {

281 + char *transaction = get_word(words, num_words, 1);

282 + const DICTIONARY_ITEM *acquired = dictionary_get_and_acquire_item(wg->worker_queue, transaction);

283 + if(acquired) {

284 + struct functions_evloop_worker_job *j = dictionary_acquired_item_value(acquired);

285 +

286 + functions_stop_monotonic_update_on_progress(&j->stop_monotonic_ut);

287 +

288 + dictionary_acquired_item_release(wg->worker_queue, acquired);

289 + }

290 + else

291 + nd_log(NDLS_COLLECTORS, NDLP_NOTICE, "Received PROGRESS for transaction '%s', but it not available here", transaction);

292 + }

293 + else if(keyword && strcmp(keyword, PLUGINSD_CALL_QUIT) == 0) {

294 + *wg->plugin_should_exit = true;

295 + return true;

296 + }

297 + else

298 + nd_log(NDLS_COLLECTORS, NDLP_NOTICE, "Received unknown command: %s", keyword ? keyword : "(unset)");

299 +

300 + buffer_flush(wg->buffer);

301 +

302 + return false;

303 +}

304 +

305 static void *rrd_functions_worker_globals_reader_main(void *arg) {

306 struct functions_evloop_globals *wg = arg;

307

193 - struct {

194 - size_t last_len; // to remember the last pos - do not use a pointer, the buffer may realloc...

195 - bool enabled;

196 - char *transaction;

197 - char *function;

198 - char *timeout_s;

199 - char *access;

200 - char *source;

201 - char *content_type;

202 - } deferred = { 0 };

203 -

204 - struct buffered_reader reader = { 0 };

205 - buffered_reader_init(&reader);

206 - BUFFER *buffer = buffer_create(sizeof(reader.read_buffer) + 2, NULL);

308 + buffered_reader_init(&wg->reader);

309 + wg->buffer = buffer_create(sizeof(wg->reader.read_buffer) + 2, NULL);

310

311 while(!(*wg->plugin_should_exit)) {

209 - if(unlikely(!buffered_reader_next_line(&reader, buffer))) {

312 + if(unlikely(!buffered_reader_next_line(&wg->reader, wg->buffer))) {

313 buffered_reader_ret_t ret = buffered_reader_read_timeout(

211 - &reader,

314 + &wg->reader,

315 fileno((FILE *)stdin),

316 2 * 60 * MSEC_PER_SEC,

317 false

@@ -220,106 +323,8 @@ static void *rrd_functions_worker_globals_reader_main(void *arg) {

323 continue;

324 }

325

223 - if(deferred.enabled) {

224 - char *s = (char *)buffer_tostring(buffer);

225 -

226 - if(strstr(&s[deferred.last_len], PLUGINSD_CALL_FUNCTION_PAYLOAD_END "\n") != NULL) {

227 - // nd_log(NDLS_COLLECTORS, NDLP_INFO, "FUNCTION PAYLOAD END");

228 -

229 - if(deferred.last_len > 0)

230 - // remove the trailing newline from the buffer

231 - deferred.last_len--;

232 -

233 - s[deferred.last_len] = '\0';

234 - buffer->len = deferred.last_len;

235 - buffer->content_type = content_type_string2id(deferred.content_type);

236 - worker_add_job(wg,

237 - PLUGINSD_CALL_FUNCTION_PAYLOAD_BEGIN, deferred.transaction, deferred.function,

238 - deferred.timeout_s, buffer, deferred.access, deferred.source);

239 - buffer_flush(buffer);

240 -

241 - freez(deferred.transaction);

242 - freez(deferred.function);

243 - freez(deferred.timeout_s);

244 - freez(deferred.access);

245 - freez(deferred.source);

246 - freez(deferred.content_type);

247 - memset(&deferred, 0, sizeof(deferred));

248 - }

249 - else

250 - deferred.last_len = buffer->len;

251 -

252 - continue;

253 - }

254 -

255 - char *words[MAX_FUNCTION_PARAMETERS] = { NULL };

256 - size_t num_words = quoted_strings_splitter_whitespace((char *)buffer_tostring(buffer), words, MAX_FUNCTION_PARAMETERS);

257 -

258 - const char *keyword = get_word(words, num_words, 0);

259 -

260 - if(keyword && (strcmp(keyword, PLUGINSD_CALL_FUNCTION) == 0)) {

261 - // nd_log(NDLS_COLLECTORS, NDLP_INFO, "FUNCTION CALL");

262 - char *transaction = get_word(words, num_words, 1);

263 - char *timeout_s = get_word(words, num_words, 2);

264 - char *function = get_word(words, num_words, 3);

265 - char *access = get_word(words, num_words, 4);

266 - char *source = get_word(words, num_words, 5);

267 - worker_add_job(wg, keyword, transaction, function, timeout_s, NULL, access, source);

268 - }

269 - else if(keyword && (strcmp(keyword, PLUGINSD_CALL_FUNCTION_PAYLOAD_BEGIN) == 0)) {

270 - // nd_log(NDLS_COLLECTORS, NDLP_INFO, "FUNCTION PAYLOAD CALL");

271 - char *transaction = get_word(words, num_words, 1);

272 - char *timeout_s = get_word(words, num_words, 2);

273 - char *function = get_word(words, num_words, 3);

274 - char *access = get_word(words, num_words, 4);

275 - char *source = get_word(words, num_words, 5);

276 - char *content_type = get_word(words, num_words, 6);

277 -

278 - deferred.transaction = strdupz(transaction ? transaction : "");

279 - deferred.timeout_s = strdupz(timeout_s ? timeout_s : "");

280 - deferred.function = strdupz(function ? function : "");

281 - deferred.access = strdupz(access ? access : "");

282 - deferred.source = strdupz(source ? source : "");

283 - deferred.content_type = strdupz(content_type ? content_type : "");

284 - deferred.last_len = 0;

285 - deferred.enabled = true;

286 - }

287 - else if(keyword && strcmp(keyword, PLUGINSD_CALL_FUNCTION_CANCEL) == 0) {

288 - // nd_log(NDLS_COLLECTORS, NDLP_INFO, "FUNCTION CANCEL");

289 - char *transaction = get_word(words, num_words, 1);

290 - const DICTIONARY_ITEM *acquired = dictionary_get_and_acquire_item(wg->worker_queue, transaction);

291 - if(acquired) {

292 - struct functions_evloop_worker_job *j = dictionary_acquired_item_value(acquired);

293 - __atomic_store_n(&j->cancelled, true, __ATOMIC_RELAXED);

294 - dictionary_acquired_item_release(wg->worker_queue, acquired);

295 - dictionary_del(wg->worker_queue, transaction);

296 - dictionary_garbage_collect(wg->worker_queue);

297 - }

298 - else

299 - nd_log(NDLS_COLLECTORS, NDLP_NOTICE, "Received CANCEL for transaction '%s', but it not available here", transaction);

300 - }

301 - else if(keyword && strcmp(keyword, PLUGINSD_CALL_FUNCTION_PROGRESS) == 0) {

302 - // nd_log(NDLS_COLLECTORS, NDLP_INFO, "FUNCTION PROGRESS");

303 - char *transaction = get_word(words, num_words, 1);

304 - const DICTIONARY_ITEM *acquired = dictionary_get_and_acquire_item(wg->worker_queue, transaction);

305 - if(acquired) {

306 - struct functions_evloop_worker_job *j = dictionary_acquired_item_value(acquired);

307 -

308 - functions_stop_monotonic_update_on_progress(&j->stop_monotonic_ut);

309 -

310 - dictionary_acquired_item_release(wg->worker_queue, acquired);

311 - }

312 - else

313 - nd_log(NDLS_COLLECTORS, NDLP_NOTICE, "Received PROGRESS for transaction '%s', but it not available here", transaction);

314 - }

315 - else if(keyword && strcmp(keyword, PLUGINSD_CALL_QUIT) == 0) {

316 - *wg->plugin_should_exit = true;

326 + if(rrd_function_worker_global_process_input(wg))

327 break;

318 - }

319 - else

320 - nd_log(NDLS_COLLECTORS, NDLP_NOTICE, "Received unknown command: %s", keyword ? keyword : "(unset)");

321 -

322 - buffer_flush(buffer);

328 }

329

330 int status = 0;

@@ -329,6 +334,7 @@ static void *rrd_functions_worker_globals_reader_main(void *arg) {

334 }

335

336 *wg->plugin_should_exit = true;

337 + buffer_free(wg->buffer);

338 exit(status);

339 }

340

src/libnetdata/libjudy/judy-malloc.c

@@ -42,6 +42,10 @@ size_t judy_aral_structures(void) {

42 return aral_structures_bytes_from_stats(&judy_sizes_aral_statistics);

43 }

45 +struct aral_statistics *judy_aral_statistics(void) {

46 + return &judy_sizes_aral_statistics;

47 +}

48 +

49 static ARAL *judy_size_aral(Word_t Words) {

50 if(Words <= MAX_JUDY_SIZE_TO_ARAL && judy_sizes_aral[Words])

51 return judy_sizes_aral[Words];

src/libnetdata/libjudy/judy-malloc.h

@@ -7,6 +7,7 @@

8 size_t judy_aral_free_bytes(void);

9 size_t judy_aral_structures(void);

10 +struct aral_statistics *judy_aral_statistics(void);

12 void JudyAllocThreadPulseReset(void);

13 int64_t JudyAllocThreadPulseGetAndReset(void);

src/ml/ml.cc

+8 -4

@@ -161,7 +161,8 @@ ml_dimension_add_model(const nd_uuid_t *metric_uuid, const ml_kmeans_inlined_t *

161 int rc = 0;

162

163 if (unlikely(!ml_db)) {

164 - error_report("Database has not been initialized");

164 + nd_log_limit_static_global_var(erl, 1, 0);

165 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR, "ML: Database has not been initialized to add ML models");

166 return 1;

167 }

168

@@ -235,7 +236,8 @@ ml_dimension_delete_models(const nd_uuid_t *metric_uuid, time_t before)

236 int param = 0;

237

238 if (unlikely(!ml_db)) {

238 - error_report("Database has not been initialized");

239 + nd_log_limit_static_global_var(erl, 1, 0);

240 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR, "ML: Database has not been initialized to delete ML models");

241 return 1;

242 }

243

@@ -285,7 +287,8 @@ ml_prune_old_models(size_t num_models_to_prune)

287 int param = 0;

288

289 if (unlikely(!ml_db)) {

288 - error_report("Database has not been initialized");

290 + nd_log_limit_static_global_var(erl, 1, 0);

291 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR, "ML: Database has not been initialized to prune old ML models");

292 return 1;

293 }

294

@@ -348,7 +351,8 @@ int ml_dimension_load_models(RRDDIM *rd, sqlite3_stmt **active_stmt) {

351 int param = 0;

352

353 if (unlikely(!ml_db)) {

351 - error_report("Database has not been initialized");

354 + nd_log_limit_static_global_var(erl, 1, 0);

355 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR, "ML: Database has not been initialized to load ML models");

356 return 1;

357 }

358

src/ml/ml_config.cc

+5 -1

@@ -45,7 +45,11 @@ void ml_config_load(ml_config_t *cfg) {

45 std::string anomaly_detection_grouping_method = config_get(config_section_ml, "anomaly detection grouping method", "average");

46 time_t anomaly_detection_query_duration = config_get_duration_seconds(config_section_ml, "anomaly detection grouping duration", 5 * 60);

48 - size_t num_worker_threads = config_get_number(config_section_ml, "num training threads", os_get_system_cpus() / 4);

48 + size_t num_worker_threads = stream_conf_configured_as_parent() ? get_netdata_cpus() / 4 : 1;

49 + if (num_worker_threads < 1) num_worker_threads = 1;

50 + else if (num_worker_threads > 256) num_worker_threads = 256;

51 + num_worker_threads = config_get_number(config_section_ml, "num training threads", num_worker_threads);

52 +

53 size_t flush_models_batch_size = config_get_number(config_section_ml, "flush models batch size", 256);

55 size_t suppression_window =

src/plugins.d/pluginsd_parser.c

+13 -11

@@ -376,13 +376,19 @@ static inline PARSER_RC pluginsd_chart(char **words, size_t num_words, PARSER *p

376 return PARSER_RC_OK;

377 }

378

379 -static void backfill_callback(size_t successful_dims __maybe_unused, size_t failed_dims __maybe_unused, struct backfill_request_data *brd) {

379 +static bool backfill_callback(size_t successful_dims __maybe_unused, size_t failed_dims __maybe_unused, struct backfill_request_data *brd) {

380 if(!rrdhost_state_acquire(brd->host, brd->rrdhost_receiver_state_id))

381 - return;

381 + return false;

382

383 - if (!replicate_chart_request(send_to_plugin, brd->parser, brd->host, brd->st,

384 - brd->first_entry_child, brd->last_entry_child, brd->child_wall_clock_time,

385 - 0, 0)) {

383 + bool rc = replicate_chart_request(send_to_plugin, brd->parser, brd->host, brd->st,

384 + brd->first_entry_child, brd->last_entry_child, brd->child_wall_clock_time,

385 + 0, 0);

386 + if (rc) {

387 + rrdset_flag_set(brd->st, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS);

388 + rrdset_flag_clear(brd->st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);

389 + rrdhost_receiver_replicating_charts_plus_one(brd->st->rrdhost);

390 + }

391 + else {

392 netdata_log_error(

393 "PLUGINSD: 'host:%s' failed to initiate replication for 'chart:%s'",

394 rrdhost_hostname(brd->host),

@@ -390,6 +396,7 @@ static void backfill_callback(size_t successful_dims __maybe_unused, size_t fail

396 }

397

398 rrdhost_state_release(brd->host);

399 + return rc;

400 }

401

402 static inline PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_words, PARSER *parser) {

@@ -416,10 +423,6 @@ static inline PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_w

423 st->replay.before = 0;

424 #endif

425

419 - rrdset_flag_set(st, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS);

420 - rrdset_flag_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);

421 - rrdhost_receiver_replicating_charts_plus_one(st->rrdhost);

422 -

426 struct backfill_request_data brd = {

427 .rrdhost_receiver_state_id = rrdhost_state_id(host),

428 .parser = parser,

@@ -432,8 +435,7 @@ static inline PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_w

435

436 ok = backfill_request_add(st, backfill_callback, &brd);

437 if(!ok)

435 - ok = replicate_chart_request(

436 - send_to_plugin, parser, host, st, first_entry_child, last_entry_child, child_wall_clock_time, 0, 0);

438 + ok = backfill_callback(0, 0, &brd);

439 }

440 #ifdef NETDATA_LOG_REPLICATION_REQUESTS

441 else {

src/streaming/replication.c

+3 -7

@@ -32,7 +32,7 @@

32 #define ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION 30

33 #define SECONDS_TO_RESET_POINT_IN_TIME 10

35 -#define MAX_REPLICATION_THREADS 32

35 +#define MAX_REPLICATION_THREADS 256

36 #define REQUESTS_AHEAD_PER_THREAD 1 // 1 = enable synchronous queries

38 static struct replication_query_statistics replication_queries = {

@@ -190,7 +190,7 @@ static struct replication_query *replication_query_prepare(

190 d->rd = rd;

191

192 STORAGE_PRIORITY priority = q->query.locked_data_collection ? STORAGE_PRIORITY_HIGH : STORAGE_PRIORITY_LOW;

193 - if(synchronous) priority = STORAGE_PRIORITY_SYNCHRONOUS;

193 + if(synchronous) priority = STORAGE_PRIORITY_SYNCHRONOUS_FIRST;

194

195 stream_control_replication_query_started();

196 storage_engine_query_init(q->backend, rd->tiers[0].smh, &d->handle,

@@ -1889,9 +1889,7 @@ void *replication_thread_main(void *ptr) {

1889

1890 replication_initialize_workers(true);

1891

1892 - int nodes = (int)dictionary_entries(rrdhost_root_index);

1893 - int cpus = (int)get_netdata_cpus();

1894 - int threads = cpus / 2;

1892 + int threads = stream_conf_configured_as_parent() ? (int)(get_netdata_cpus() / 2) : 1;

1893 if (threads < 1) threads = 1;

1894 else if (threads > MAX_REPLICATION_THREADS) threads = MAX_REPLICATION_THREADS;

1895

@@ -1907,8 +1905,6 @@ void *replication_thread_main(void *ptr) {

1905 config_set_number(CONFIG_SECTION_DB, "replication threads", threads);

1906 }

1907

1910 - netdata_log_info("replication threads set to %d (cpu cores = %d, nodes = %d)", threads, cpus, nodes);

1911 -

1908 if(--threads) {

1909 replication_globals.main_thread.threads = threads;

1910 replication_globals.main_thread.threads_ptrs = mallocz(threads * sizeof(ND_THREAD *));

src/streaming/stream-connector.c

-4

@@ -549,10 +549,6 @@ void stream_connector_add(struct sender_state *s) {

549 }

550

551 static void stream_connector_remove(struct sender_state *s) {

552 - nd_log(NDLS_DAEMON, NDLP_NOTICE,

553 - "STREAM CONNECT '%s' [stopped]: stopped streaming connector for host, reason: %s",

554 - rrdhost_hostname(s->host), stream_handshake_error_to_string(s->exit.reason));

555 -

552 struct connector *sc = stream_connector_get(s);

553 __atomic_sub_fetch(&sc->nodes, 1, __ATOMIC_RELAXED);

554

src/streaming/stream-receiver.c

+6 -11

@@ -466,14 +466,18 @@ void stream_receiver_move_entire_queue_to_running_unsafe(struct stream_thread *s

466 static void stream_receiver_remove(struct stream_thread *sth, struct receiver_state *rpt, const char *why) {

467 internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ );

468

469 + PARSER *parser = __atomic_load_n(&rpt->thread.parser, __ATOMIC_RELAXED);

470 + size_t count = parser ? parser->user.data_collections_count : 0;

471 +

472 errno_clear();

473 nd_log(NDLS_DAEMON, NDLP_ERR,

474 "STREAM RCV[%zu] '%s' [from [%s]:%s]: "

472 - "receiver disconnected: %s"

475 + "receiver disconnected (after %zu received messages): %s"

476 , sth->id

477 , rpt->hostname ? rpt->hostname : "-"

478 , rpt->client_ip ? rpt->client_ip : "-"

479 , rpt->client_port ? rpt->client_port : "-"

480 + , count

481 , why ? why : "");

482

483 rrdhost_state_disconnected(rpt->host);

@@ -498,8 +502,6 @@ static void stream_receiver_remove(struct stream_thread *sth, struct receiver_st

502 buffer_free(rpt->thread.buffer);

503 rpt->thread.buffer = NULL;

504

501 - size_t count = 0;

502 - PARSER *parser = __atomic_load_n(&rpt->thread.parser, __ATOMIC_RELAXED);

505 if(parser) {

506 parser->user.v2.stream_buffer.wb = NULL;

507

@@ -509,19 +511,11 @@ static void stream_receiver_remove(struct stream_thread *sth, struct receiver_st

511 parser->fd_output = -1;

512 parser->sock = NULL;

513 spinlock_unlock(&parser->writer.spinlock);

512 -

513 - count = parser->user.data_collections_count;

514 }

515

516 // the parser stopped

517 receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_PARSER_EXIT, false);

518

519 - {

520 - char msg[100 + 1];

521 - snprintfz(msg, sizeof(msg) - 1, "receiver disconnected (completed %zu updates)", count);

522 - stream_receiver_log_status(rpt, msg, STREAM_STATUS_DISCONNECTED, NDLP_WARNING);

523 - }

524 -

519 // in case we are connected to netdata cloud,

520 // we inform cloud that a child got disconnected

521 uint64_t total_reboot = rrdhost_stream_path_total_reboot_time_ms(rpt->host);

@@ -786,6 +780,7 @@ void stream_receiver_cleanup(struct stream_thread *sth) {

780 m = META_NEXT(&sth->run.meta, &idx)) {

781 if (m->type != POLLFD_TYPE_RECEIVER) continue;

782 struct receiver_state *rpt = m->rpt;

783 + receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_SHUTDOWN, false);

784 stream_receiver_remove(sth, rpt, "shutdown");

785 }

786 }

src/streaming/stream-sender.c

@@ -619,6 +619,8 @@ void stream_sender_cleanup(struct stream_thread *sth) {

619 };

620 ND_LOG_STACK_PUSH(lgs);

621

622 + s->exit.reason = STREAM_HANDSHAKE_DISCONNECT_SHUTDOWN;

623 + s->exit.shutdown = true;

624 stream_sender_move_running_to_connector_or_remove(sth, s, STREAM_HANDSHAKE_DISCONNECT_SHUTDOWN, false);

625 }

626 }

src/web/api/queries/backfill.h

+1 -1

@@ -16,7 +16,7 @@ struct backfill_request_data {

16 time_t child_wall_clock_time;

17 };

19 -typedef void (*backfill_callback_t)(size_t successful_dims, size_t failed_dims, struct backfill_request_data *brd);

19 +typedef bool (*backfill_callback_t)(size_t successful_dims, size_t failed_dims, struct backfill_request_data *brd);

21 void *backfill_thread(void *ptr);

22 bool backfill_request_add(RRDSET *st, backfill_callback_t cb, struct backfill_request_data *data);

src/web/api/queries/query.c

+1 -1

@@ -2002,7 +2002,7 @@ bool backfill_tier_from_smaller_tiers(RRDDIM *rd, size_t tier, time_t now_s) {

2002 long before_wanted = smaller_tier_last_time;

2003

2004 struct rrddim_tier *tmp = &rd->tiers[read_tier];

2005 - storage_engine_query_init(tmp->seb, tmp->smh, &seqh, after_wanted, before_wanted, STORAGE_PRIORITY_SYNCHRONOUS);

2005 + storage_engine_query_init(tmp->seb, tmp->smh, &seqh, after_wanted, before_wanted, STORAGE_PRIORITY_SYNCHRONOUS_FIRST);

2006

2007 size_t points_read = 0;

2008

src/web/api/queries/weights.c

+5 -5

@@ -1281,7 +1281,7 @@ NETDATA_DOUBLE *rrd2rrdr_ks2(

1281 .time_group_options = time_group_options,

1282 .tier = tier,

1283 .query_source = QUERY_SOURCE_API_WEIGHTS,

1284 - .priority = STORAGE_PRIORITY_SYNCHRONOUS,

1284 + .priority = STORAGE_PRIORITY_SYNCHRONOUS_FIRST,

1285 };

1286

1287 QUERY_TARGET *qt = query_target_create(&qtr);

@@ -1420,7 +1420,7 @@ static void rrdset_metric_correlations_volume(

1420

1421 QUERY_VALUE baseline_average = rrdmetric2value(host, rca, ria, rma, baseline_after, baseline_before,

1422 options, time_group_method, time_group_options, tier, 0,

1423 - QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS);

1423 + QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS_FIRST);

1424 merge_query_value_to_stats(&baseline_average, stats, 1);

1425

1426 if(!netdata_double_isnumber(baseline_average.value)) {

@@ -1430,7 +1430,7 @@ static void rrdset_metric_correlations_volume(

1430

1431 QUERY_VALUE highlight_average = rrdmetric2value(host, rca, ria, rma, after, before,

1432 options, time_group_method, time_group_options, tier, 0,

1433 - QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS);

1433 + QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS_FIRST);

1434 merge_query_value_to_stats(&highlight_average, stats, 1);

1435

1436 if(!netdata_double_isnumber(highlight_average.value))

@@ -1450,7 +1450,7 @@ static void rrdset_metric_correlations_volume(

1450 snprintfz(highlight_countif_options, 50, "%s" NETDATA_DOUBLE_FORMAT, highlight_average.value < baseline_average.value ? "<" : ">", baseline_average.value);

1451 QUERY_VALUE highlight_countif = rrdmetric2value(host, rca, ria, rma, after, before,

1452 options, RRDR_GROUPING_COUNTIF, highlight_countif_options, tier, 0,

1453 - QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS);

1453 + QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS_FIRST);

1454 merge_query_value_to_stats(&highlight_countif, stats, 1);

1455

1456 if(!netdata_double_isnumber(highlight_countif.value)) {

@@ -1494,7 +1494,7 @@ static void rrdset_weights_value(

1494

1495 QUERY_VALUE qv = rrdmetric2value(host, rca, ria, rma, after, before,

1496 options, time_group_method, time_group_options, tier, 0,

1497 - QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS);

1497 + QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS_FIRST);

1498

1499 merge_query_value_to_stats(&qv, stats, 1);

1500