@cryptotaxi247 / netdata-1 / commits / c1908d316

DBENGINE v2 - improvements part 5 (#14289)

* cleanup journal v2 mounts periodically * fix for last commit * re-enable loading page from disk when the arrangement of pages requires it * Remove unused statistics * Estimate diskspace when the current datafile is full and queue a rotate command (Currently it will not attempt to estimate end size for journals) Queue a command to check quota on startup per tier * apps.plugin now exposes RSS chart * shorter thread names to make debugging easier, since thread names can only be 15 characters * more thread names fixes * allow an apps_groups.conf target to be pid 0 or 1 Co-authored-by: Stelios Fragkakis <52996999+stelfrag@users.noreply.github.com>

Costa Tsaousis committed Jan 18, 2023 at 21:32 UTC c1908d3163185cf65a139edeb11a165a10eca1e9
29 files changed +214 -147
aclk/aclk.c
+2 -2
@@ -731,8 +731,8 @@ void *aclk_main(void *ptr)
731 stats_thread->client = mqttwss_client;
732 aclk_stats_thread_prepare(query_threads.count, proto_hdl_cnt);
733 netdata_thread_create(
734 - stats_thread->thread, ACLK_STATS_THREAD_NAME, NETDATA_THREAD_OPTION_JOINABLE, aclk_stats_main_thread,
735 - stats_thread);
734 + stats_thread->thread, "ACLK_STATS", NETDATA_THREAD_OPTION_JOINABLE, aclk_stats_main_thread,
735 + stats_thread);
736 }
737
738 // Keep reconnecting and talking until our time has come
aclk/aclk_query.c
+1 -3
@@ -4,8 +4,6 @@
4 #include "aclk_stats.h"
5 #include "aclk_tx_msgs.h"
6
7 -#define ACLK_QUERY_THREAD_NAME "ACLK_Query"
8 -
7 #define WEB_HDR_ACCEPT_ENC "Accept-Encoding:"
8
9 pthread_cond_t query_cond_wait = PTHREAD_COND_INITIALIZER;
@@ -368,7 +366,7 @@ void aclk_query_threads_start(struct aclk_query_threads *query_threads, mqtt_wss
366 query_threads->thread_list[i].idx = i; //thread needs to know its index for statistics
367 query_threads->thread_list[i].client = client;
368
371 - if(unlikely(snprintfz(thread_name, TASK_LEN_MAX, "%s_%d", ACLK_QUERY_THREAD_NAME, i) < 0))
369 + if(unlikely(snprintfz(thread_name, TASK_LEN_MAX, "ACLK_QRY[%d]", i) < 0))
370 error("snprintf encoding error");
371 netdata_thread_create(
372 &query_threads->thread_list[i].thread, thread_name, NETDATA_THREAD_OPTION_JOINABLE, aclk_query_main_thread,
aclk/aclk_stats.h
-2
@@ -8,8 +8,6 @@
8 #include "aclk_query_queue.h"
9 #include "mqtt_wss_client.h"
10
11 -#define ACLK_STATS_THREAD_NAME "ACLK_Stats"
12 -
11 extern netdata_mutex_t aclk_stats_mutex;
12
13 #define ACLK_STATS_LOCK netdata_mutex_lock(&aclk_stats_mutex)
collectors/apps.plugin/apps_plugin.c
+17 -2
@@ -420,6 +420,7 @@ struct pid_stat {
420 int sortlist; // higher numbers = top on the process tree
421 // each process gets a unique number
422
423 + bool matched_by_config;
424 struct target *target; // app_groups.conf targets
425 struct target *user_target; // uid based targets
426 struct target *group_target; // gid based targets
@@ -1103,6 +1104,7 @@ static inline void assign_target_to_pid(struct pid_stat *p) {
1104 || (proc_pid_cmdline_is_needed && w->starts_with && w->ends_with && p->cmdline && strstr(p->cmdline, w->compare))
1105 ))) {
1106
1107 + p->matched_by_config = true;
1108 if(w->target) p->target = w->target;
1109 else p->target = w;
1110
@@ -2832,11 +2834,11 @@ static void apply_apps_groups_targets_inheritance(void) {
2834 }
2835
2836 // init goes always to default target
2835 - if(all_pids[INIT_PID])
2837 + if(all_pids[INIT_PID] && !all_pids[INIT_PID]->matched_by_config)
2838 all_pids[INIT_PID]->target = apps_groups_default_target;
2839
2840 // pid 0 goes always to default target
2839 - if(all_pids[0])
2841 + if(all_pids[0] && !all_pids[INIT_PID]->matched_by_config)
2842 all_pids[0]->target = apps_groups_default_target;
2843
2844 // give a default target on all top level processes
@@ -3589,6 +3591,13 @@ static void send_collected_data_to_netdata(struct target *root, const char *type
3591 }
3592 send_END();
3593
3594 + send_BEGIN(type, "rss", dt);
3595 + for (w = root; w ; w = w->next) {
3596 + if(unlikely(w->exposed && w->processes))
3597 + send_SET(w->name, w->status_vmrss);
3598 + }
3599 + send_END();
3600 +
3601 send_BEGIN(type, "vmem", dt);
3602 for (w = root; w ; w = w->next) {
3603 if(unlikely(w->exposed && w->processes))
@@ -3728,6 +3737,12 @@ static void send_charts_updates_to_netdata(struct target *root, const char *type
3737 }
3738 APPS_PLUGIN_FUNCTIONS();
3739
3740 + fprintf(stdout, "CHART %s.rss '' '%s Resident Set Size (w/shared)' 'MiB' mem %s.rss stacked 20004 %d\n", type, title, type, update_every);
3741 + for (w = root; w ; w = w->next) {
3742 + if(unlikely(w->exposed))
3743 + fprintf(stdout, "DIMENSION %s '' absolute %ld %ld\n", w->name, 1L, 1024L);
3744 + }
3745 + APPS_PLUGIN_FUNCTIONS();
3746
3747 fprintf(stdout, "CHART %s.vmem '' '%s Virtual Memory Size' 'MiB' mem %s.vmem stacked 20005 %d\n", type, title, type, update_every);
3748 for (w = root; w ; w = w->next) {
collectors/diskspace.plugin/plugin_diskspace.c
+1 -2
@@ -3,7 +3,6 @@
3 #include "../proc.plugin/plugin_proc.h"
4
5 #define PLUGIN_DISKSPACE_NAME "diskspace.plugin"
6 -#define THREAD_DISKSPACE_SLOW_NAME "PLUGIN[diskspace slow]"
6
7 #define DEFAULT_EXCLUDED_PATHS "/proc/* /sys/* /var/run/user/* /run/user/* /snap/* /var/lib/docker/*"
8 #define DEFAULT_EXCLUDED_FILESYSTEMS "*gvfs *gluster* *s3fs *ipfs *davfs2 *httpfs *sshfs *gdfs *moosefs fusectl autofs"
@@ -632,7 +631,7 @@ void *diskspace_main(void *ptr) {
631
632 netdata_thread_create(
633 diskspace_slow_thread,
635 - THREAD_DISKSPACE_SLOW_NAME,
634 + "P[diskspace slow]",
635 NETDATA_THREAD_OPTION_JOINABLE,
636 diskspace_slow_worker,
637 &slow_worker_data);
collectors/plugins.d/plugins_d.c
+1 -1
@@ -271,7 +271,7 @@ void *pluginsd_main(void *ptr)
271
272 if (cd->enabled) {
273 char tag[NETDATA_THREAD_TAG_MAX + 1];
274 - snprintfz(tag, NETDATA_THREAD_TAG_MAX, "PLUGINSD[%s]", pluginname);
274 + snprintfz(tag, NETDATA_THREAD_TAG_MAX, "PD[%s]", pluginname);
275 // spawn a new thread for it
276 netdata_thread_create(
277 &cd->thread, tag, NETDATA_THREAD_OPTION_DEFAULT, pluginsd_worker_thread, cd);
collectors/proc.plugin/plugin_proc.h
+1 -1
@@ -8,7 +8,7 @@
8 #define PLUGIN_PROC_CONFIG_NAME "proc"
9 #define PLUGIN_PROC_NAME PLUGIN_PROC_CONFIG_NAME ".plugin"
10
11 -#define THREAD_NETDEV_NAME "PLUGIN[proc netdev]"
11 +#define THREAD_NETDEV_NAME "P[proc netdev]"
12 void *netdev_main(void *ptr);
13
14 int do_proc_net_wireless(int update_every, usec_t dt);
collectors/statsd.plugin/statsd.c
+1 -1
@@ -2548,7 +2548,7 @@ void *statsd_main(void *ptr) {
2548 for(i = 0; i < statsd.threads ;i++) {
2549 statsd.collection_threads_status[i].max_sockets = max_sockets / statsd.threads;
2550 char tag[NETDATA_THREAD_TAG_MAX + 1];
2551 - snprintfz(tag, NETDATA_THREAD_TAG_MAX, "STATSD_COLLECTOR[%d]", i + 1);
2551 + snprintfz(tag, NETDATA_THREAD_TAG_MAX, "STATSD_IN[%d]", i + 1);
2552 netdata_spinlock_init(&statsd.collection_threads_status[i].spinlock);
2553 netdata_thread_create(&statsd.collection_threads_status[i].thread, tag, NETDATA_THREAD_OPTION_DEFAULT, statsd_collector_thread, &statsd.collection_threads_status[i]);
2554 }
daemon/static_threads.c
+16 -16
@@ -19,7 +19,7 @@ extern bool global_statistics_enabled;
19
20 const struct netdata_static_thread static_threads_common[] = {
21 {
22 - .name = "PLUGIN[timex]",
22 + .name = "P[timex]",
23 .config_section = CONFIG_SECTION_PLUGINS,
24 .config_name = "timex",
25 .enabled = 1,
@@ -28,7 +28,7 @@ const struct netdata_static_thread static_threads_common[] = {
28 .start_routine = timex_main
29 },
30 {
31 - .name = "PLUGIN[idlejitter]",
31 + .name = "P[idlejitter]",
32 .config_section = CONFIG_SECTION_PLUGINS,
33 .config_name = "idlejitter",
34 .enabled = 1,
@@ -55,7 +55,7 @@ const struct netdata_static_thread static_threads_common[] = {
55 .start_routine = analytics_main
56 },
57 {
58 - .name = "GLOBAL_STATS",
58 + .name = "STATS_GLOBAL",
59 .config_section = CONFIG_SECTION_PLUGINS,
60 .config_name = "netdata monitoring",
61 .env_name = "NETDATA_INTERNALS_MONITORING",
@@ -66,7 +66,7 @@ const struct netdata_static_thread static_threads_common[] = {
66 .start_routine = global_statistics_main
67 },
68 {
69 - .name = "WORKERS_STATS",
69 + .name = "STATS_WORKERS",
70 .config_section = CONFIG_SECTION_PLUGINS,
71 .config_name = "netdata monitoring",
72 .env_name = "NETDATA_INTERNALS_MONITORING",
@@ -77,7 +77,7 @@ const struct netdata_static_thread static_threads_common[] = {
77 .start_routine = global_statistics_workers_main
78 },
79 {
80 - .name = "SQLITE3_STATS",
80 + .name = "STATS_SQLITE3",
81 .config_section = CONFIG_SECTION_PLUGINS,
82 .config_name = "netdata monitoring",
83 .env_name = "NETDATA_INTERNALS_MONITORING",
@@ -106,7 +106,7 @@ const struct netdata_static_thread static_threads_common[] = {
106 .start_routine = service_main
107 },
108 {
109 - .name = "STATSD",
109 + .name = "STATSD_FLUSH",
110 .config_section = NULL,
111 .config_name = NULL,
112 .enabled = 1,
@@ -124,7 +124,7 @@ const struct netdata_static_thread static_threads_common[] = {
124 .start_routine = exporting_main
125 },
126 {
127 - .name = "STREAM",
127 + .name = "SNDR[localhost]",
128 .config_section = NULL,
129 .config_name = NULL,
130 .enabled = 0,
@@ -133,7 +133,7 @@ const struct netdata_static_thread static_threads_common[] = {
133 .start_routine = rrdpush_sender_thread
134 },
135 {
136 - .name = "WEB_SERVER[static1]",
136 + .name = "WEB[1]",
137 .config_section = NULL,
138 .config_name = NULL,
139 .enabled = 0,
@@ -144,7 +144,7 @@ const struct netdata_static_thread static_threads_common[] = {
144
145 #ifdef ENABLE_ACLK
146 {
147 - .name = "ACLK_Main",
147 + .name = "ACLK_MAIN",
148 .config_section = NULL,
149 .config_name = NULL,
150 .enabled = 1,
@@ -165,13 +165,13 @@ const struct netdata_static_thread static_threads_common[] = {
165 },
166
167 {
168 - .name = "REPLICATION",
169 - .config_section = NULL,
170 - .config_name = NULL,
171 - .enabled = 1,
172 - .thread = NULL,
173 - .init_routine = NULL,
174 - .start_routine = replication_thread_main
168 + .name = "REPLAY[1]",
169 + .config_section = NULL,
170 + .config_name = NULL,
171 + .enabled = 1,
172 + .thread = NULL,
173 + .init_routine = NULL,
174 + .start_routine = replication_thread_main
175 },
176
177 // terminator
daemon/static_threads_freebsd.c
+1 -1
@@ -6,7 +6,7 @@ extern void *freebsd_main(void *ptr);
6
7 const struct netdata_static_thread static_threads_freebsd[] = {
8 {
9 - .name = "PLUGIN[freebsd]",
9 + .name = "P[freebsd]",
10 .config_section = CONFIG_SECTION_PLUGINS,
11 .config_name = "freebsd",
12 .enabled = 1,
daemon/static_threads_linux.c
+4 -4
@@ -10,7 +10,7 @@ extern void *timex_main(void *ptr);
10
11 const struct netdata_static_thread static_threads_linux[] = {
12 {
13 - .name = "PLUGIN[tc]",
13 + .name = "P[tc]",
14 .config_section = CONFIG_SECTION_PLUGINS,
15 .config_name = "tc",
16 .enabled = 1,
@@ -19,7 +19,7 @@ const struct netdata_static_thread static_threads_linux[] = {
19 .start_routine = tc_main
20 },
21 {
22 - .name = "PLUGIN[diskspace]",
22 + .name = "P[diskspace]",
23 .config_section = CONFIG_SECTION_PLUGINS,
24 .config_name = "diskspace",
25 .enabled = 1,
@@ -28,7 +28,7 @@ const struct netdata_static_thread static_threads_linux[] = {
28 .start_routine = diskspace_main
29 },
30 {
31 - .name = "PLUGIN[proc]",
31 + .name = "P[proc]",
32 .config_section = CONFIG_SECTION_PLUGINS,
33 .config_name = "proc",
34 .enabled = 1,
@@ -37,7 +37,7 @@ const struct netdata_static_thread static_threads_linux[] = {
37 .start_routine = proc_main
38 },
39 {
40 - .name = "PLUGIN[cgroups]",
40 + .name = "P[cgroups]",
41 .config_section = CONFIG_SECTION_PLUGINS,
42 .config_name = "cgroups",
43 .enabled = 1,
daemon/static_threads_macos.c
+1 -1
@@ -6,7 +6,7 @@ extern void *macos_main(void *ptr);
6
7 const struct netdata_static_thread static_threads_macos[] = {
8 {
9 - .name = "PLUGIN[macos]",
9 + .name = "P[macos]",
10 .config_section = CONFIG_SECTION_PLUGINS,
11 .config_name = "macos",
12 .enabled = 1,
database/engine/datafile.c
+2 -2
@@ -460,7 +460,7 @@ static int scan_data_files(struct rrdengine_instance *ctx)
460 error("DBENGINE: deleting invalid data and journal file pair.");
461 ret = journalfile_unlink(journalfile);
462 if (!ret) {
463 - journalfile_generate_path(datafile, path, sizeof(path));
463 + journalfile_v1_generate_path(datafile, path, sizeof(path));
464 info("DBENGINE: deleted journal file \"%s\".", path);
465 }
466 ret = unlink_data_file(datafile);
@@ -508,7 +508,7 @@ int create_new_datafile_pair(struct rrdengine_instance *ctx)
508 if (ret)
509 goto error_after_journalfile;
510
511 - journalfile_generate_path(datafile, path, sizeof(path));
511 + journalfile_v1_generate_path(datafile, path, sizeof(path));
512 info("DBENGINE: created journal file \"%s\".", path);
513
514 datafile_list_insert(ctx, datafile);
database/engine/journalfile.c
+104 -75
@@ -113,7 +113,7 @@ void journalfile_v2_generate_path(struct rrdengine_datafile *datafile, char *str
113 datafile->ctx->dbfiles_path, datafile->tier, datafile->fileno);
114 }
115
116 -void journalfile_generate_path(struct rrdengine_datafile *datafile, char *str, size_t maxlen)
116 +void journalfile_v1_generate_path(struct rrdengine_datafile *datafile, char *str, size_t maxlen)
117 {
118 (void) snprintfz(str, maxlen, "%s/" WALFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL WALFILE_EXTENSION,
119 datafile->ctx->dbfiles_path, datafile->tier, datafile->fileno);
@@ -166,12 +166,25 @@ static struct journal_v2_header *journalfile_v2_mounted_data_get(struct rrdengin
166 return j2_header;
167 }
168
169 -static bool journalfile_v2_mounted_data_unmount(struct rrdengine_journalfile *journalfile, bool have_locks) {
169 +static bool journalfile_v2_mounted_data_unmount(struct rrdengine_journalfile *journalfile, bool have_locks, bool wait) {
170 bool unmounted = false;
171
172 if(!have_locks) {
173 - netdata_spinlock_lock(&journalfile->mmap.spinlock);
174 - netdata_spinlock_lock(&journalfile->v2.spinlock);
173 + if(!wait) {
174 + if (!netdata_spinlock_trylock(&journalfile->mmap.spinlock))
175 + return false;
176 + }
177 + else
178 + netdata_spinlock_lock(&journalfile->mmap.spinlock);
179 +
180 + if(!wait) {
181 + if(!netdata_spinlock_trylock(&journalfile->v2.spinlock)) {
182 + netdata_spinlock_unlock(&journalfile->mmap.spinlock);
183 + return false;
184 + }
185 + }
186 + else
187 + netdata_spinlock_lock(&journalfile->v2.spinlock);
188 }
189
190 if(!journalfile->v2.refcount) {
@@ -202,13 +215,49 @@ static bool journalfile_v2_mounted_data_unmount(struct rrdengine_journalfile *jo
215 return unmounted;
216 }
217
218 +void journalfile_v2_data_unmount_cleanup(time_t now_s) {
219 + // DO NOT WAIT ON ANY LOCK!!!
220 +
221 + for(size_t tier = 0; tier < RRD_STORAGE_TIERS ;tier++) {
222 + struct rrdengine_instance *ctx = multidb_ctx[tier];
223 + if(!ctx) continue;
224 +
225 + struct rrdengine_datafile *datafile;
226 + if(uv_rwlock_tryrdlock(&ctx->datafiles.rwlock) != 0)
227 + continue;
228 +
229 + for (datafile = ctx->datafiles.first; datafile; datafile = datafile->next) {
230 + struct rrdengine_journalfile *journalfile = datafile->journalfile;
231 +
232 + if(!netdata_spinlock_trylock(&journalfile->v2.spinlock))
233 + continue;
234 +
235 + bool unmount = false;
236 + if (!journalfile->v2.refcount && (journalfile->v2.flags & JOURNALFILE_FLAG_IS_MOUNTED)) {
237 + // this journal has no references and it is mounted
238 +
239 + if (!journalfile->v2.not_needed_since_s)
240 + journalfile->v2.not_needed_since_s = now_s;
241 +
242 + else if (now_s - journalfile->v2.not_needed_since_s >= 120)
243 + // 2 minutes have passed since last use
244 + unmount = true;
245 + }
246 + netdata_spinlock_unlock(&journalfile->v2.spinlock);
247 +
248 + if (unmount)
249 + journalfile_v2_mounted_data_unmount(journalfile, false, false);
250 + }
251 + uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
252 + }
253 +}
254 +
255 struct journal_v2_header *journalfile_v2_data_acquire(struct rrdengine_journalfile *journalfile, size_t *data_size, time_t wanted_first_time_s, time_t wanted_last_time_s) {
256 netdata_spinlock_lock(&journalfile->v2.spinlock);
257
258 bool has_data = (journalfile->v2.flags & JOURNALFILE_FLAG_IS_AVAILABLE);
259 bool is_mounted = (journalfile->v2.flags & JOURNALFILE_FLAG_IS_MOUNTED);
260 bool do_we_need_it = false;
211 - bool unmount = false;
261
262 if(has_data) {
263 if (!wanted_first_time_s || !wanted_last_time_s ||
@@ -218,7 +267,6 @@ struct journal_v2_header *journalfile_v2_data_acquire(struct rrdengine_journalfi
267 journalfile->v2.refcount++;
268
269 do_we_need_it = true;
221 - journalfile->v2.not_needed_counter = 0;
270
271 if (!wanted_first_time_s && !wanted_last_time_s && !is_mounted)
272 journalfile->v2.flags |= JOURNALFILE_FLAG_MOUNTED_FOR_RETENTION;
@@ -226,33 +274,12 @@ struct journal_v2_header *journalfile_v2_data_acquire(struct rrdengine_journalfi
274 journalfile->v2.flags &= ~JOURNALFILE_FLAG_MOUNTED_FOR_RETENTION;
275
276 }
229 - else if (is_mounted) {
230 - // this journal has data, but it does not match our query
231 -
232 - if (!journalfile->v2.refcount) {
233 - // this journal has no references
234 -
235 - if (!journalfile->v2.not_needed_counter)
236 - journalfile->v2.not_needed_since_s = now_monotonic_sec();
237 -
238 - if ((++journalfile->v2.not_needed_counter) % 100 == 0) {
239 - // at least 100 times it has been evaluated since last use
240 -
241 - if (now_monotonic_sec() - journalfile->v2.not_needed_since_s >= 120)
242 - // 2 minutes have passed since last use
243 - unmount = true;
244 - }
245 - }
246 - }
277 }
278 netdata_spinlock_unlock(&journalfile->v2.spinlock);
279
280 if(do_we_need_it)
281 return journalfile_v2_mounted_data_get(journalfile, data_size);
282
253 - else if(unmount)
254 - journalfile_v2_mounted_data_unmount(journalfile, false);
255 -
283 return NULL;
284 }
285
@@ -267,7 +294,7 @@ void journalfile_v2_data_release(struct rrdengine_journalfile *journalfile) {
294 journalfile->v2.refcount--;
295
296 if(journalfile->v2.refcount == 0) {
270 - journalfile->v2.not_needed_counter = 0;
297 + journalfile->v2.not_needed_since_s = 0;
298
299 if(journalfile->v2.flags & JOURNALFILE_FLAG_MOUNTED_FOR_RETENTION)
300 unmount = true;
@@ -275,7 +302,7 @@ void journalfile_v2_data_release(struct rrdengine_journalfile *journalfile) {
302 netdata_spinlock_unlock(&journalfile->v2.spinlock);
303
304 if(unmount)
278 - journalfile_v2_mounted_data_unmount(journalfile, false);
305 + journalfile_v2_mounted_data_unmount(journalfile, false, true);
306 }
307
308 bool journalfile_v2_data_available(struct rrdengine_journalfile *journalfile) {
@@ -314,7 +341,7 @@ void journalfile_v2_data_set(struct rrdengine_journalfile *journalfile, int fd,
341 journalfile->v2.first_time_s = (time_t)(j2_header->start_time_ut / USEC_PER_SEC);
342 journalfile->v2.last_time_s = (time_t)(j2_header->end_time_ut / USEC_PER_SEC);
343
317 - journalfile_v2_mounted_data_unmount(journalfile, true);
344 + journalfile_v2_mounted_data_unmount(journalfile, true, true);
345
346 netdata_spinlock_unlock(&journalfile->v2.spinlock);
347 netdata_spinlock_unlock(&journalfile->mmap.spinlock);
@@ -330,8 +357,10 @@ static void journalfile_v2_data_unmap_permanently(struct rrdengine_journalfile *
357 netdata_spinlock_lock(&journalfile->mmap.spinlock);
358 netdata_spinlock_lock(&journalfile->v2.spinlock);
359
333 - if(journalfile_v2_mounted_data_unmount(journalfile, true)) {
334 - close(journalfile->mmap.fd);
360 + if(journalfile_v2_mounted_data_unmount(journalfile, true, true)) {
361 + if(journalfile->mmap.fd != -1)
362 + close(journalfile->mmap.fd);
363 +
364 journalfile->mmap.fd = -1;
365 journalfile->mmap.data = NULL;
366 journalfile->mmap.size = 0;
@@ -369,7 +398,7 @@ static int close_uv_file(struct rrdengine_datafile *datafile, uv_file file)
398 uv_fs_t req;
399 ret = uv_fs_close(NULL, &req, file, NULL);
400 if (ret < 0) {
372 - journalfile_generate_path(datafile, path, sizeof(path));
401 + journalfile_v1_generate_path(datafile, path, sizeof(path));
402 error("DBENGINE: uv_fs_close(%s): %s", path, uv_strerror(ret));
403 ++datafile->ctx->stats.fs_errors;
404 rrd_stat_atomic_add(&global_fs_errors, 1);
@@ -396,7 +425,7 @@ int journalfile_unlink(struct rrdengine_journalfile *journalfile)
425 int ret;
426 char path[RRDENG_PATH_MAX];
427
399 - journalfile_generate_path(datafile, path, sizeof(path));
428 + journalfile_v1_generate_path(datafile, path, sizeof(path));
429
430 ret = uv_fs_unlink(NULL, &req, path, NULL);
431 if (ret < 0) {
@@ -419,7 +448,7 @@ int journalfile_destroy_unsafe(struct rrdengine_journalfile *journalfile, struct
448 char path[RRDENG_PATH_MAX];
449 char path_v2[RRDENG_PATH_MAX];
450
422 - journalfile_generate_path(datafile, path, sizeof(path));
451 + journalfile_v1_generate_path(datafile, path, sizeof(path));
452 journalfile_v2_generate_path(datafile, path_v2, sizeof(path));
453
454 if (journalfile->file) {
@@ -469,7 +498,7 @@ int journalfile_create(struct rrdengine_journalfile *journalfile, struct rrdengi
498 uv_buf_t iov;
499 char path[RRDENG_PATH_MAX];
500
472 - journalfile_generate_path(datafile, path, sizeof(path));
501 + journalfile_v1_generate_path(datafile, path, sizeof(path));
502 fd = open_file_direct_io(path, O_CREAT | O_RDWR | O_TRUNC, &file);
503 if (fd < 0) {
504 ++ctx->stats.fs_errors;
@@ -783,7 +812,7 @@ static int journalfile_check_v2_metric_list(void *data_start, size_t file_size)
812 // 2 Force rebuild
813 // 3 skip
814
786 -static int journalfile_v2_validate(void *data_start, size_t file_size, uint32_t original_size)
815 +static int journalfile_v2_validate(void *data_start, size_t journal_v2_file_size, size_t journal_v1_file_size)
816 {
817 int rc;
818 uLong crc;
@@ -801,13 +830,13 @@ static int journalfile_v2_validate(void *data_start, size_t file_size, uint32_t
830 if (j2_header->magic != JOURVAL_V2_MAGIC)
831 return 1;
832
804 - if (j2_header->total_file_size != file_size)
833 + if (j2_header->journal_v2_file_size != journal_v2_file_size)
834 return 1;
835
807 - if (original_size && j2_header->original_file_size != original_size)
836 + if (journal_v1_file_size && j2_header->journal_v1_file_size != journal_v1_file_size)
837 return 1;
838
810 - journal_v2_trailer = (struct journal_v2_block_trailer *) ((uint8_t *) data_start + file_size - sizeof(*journal_v2_trailer));
839 + journal_v2_trailer = (struct journal_v2_block_trailer *) ((uint8_t *) data_start + journal_v2_file_size - sizeof(*journal_v2_trailer));
840
841 crc = crc32(0L, Z_NULL, 0);
842 crc = crc32(crc, (void *) j2_header, sizeof(*j2_header));
@@ -818,10 +847,10 @@ static int journalfile_v2_validate(void *data_start, size_t file_size, uint32_t
847 return 1;
848 }
849
821 - rc = journalfile_check_v2_extent_list(data_start, file_size);
850 + rc = journalfile_check_v2_extent_list(data_start, journal_v2_file_size);
851 if (rc) return 1;
852
824 - rc = journalfile_check_v2_metric_list(data_start, file_size);
853 + rc = journalfile_check_v2_metric_list(data_start, journal_v2_file_size);
854 if (rc) return 1;
855
856 if (!db_engine_journal_check)
@@ -865,7 +894,7 @@ static int journalfile_v2_validate(void *data_start, size_t file_size, uint32_t
894 }
895
896 metric++;
868 - if ((uint32_t)((uint8_t *) metric - (uint8_t *) data_start) > (uint32_t) file_size) {
897 + if ((uint32_t)((uint8_t *) metric - (uint8_t *) data_start) > (uint32_t) journal_v2_file_size) {
898 info("DBENGINE: verification failed EOF reached -- total entries %u, verified %u", entries, verified);
899 return 1;
900 }
@@ -883,62 +912,62 @@ static int journalfile_v2_validate(void *data_start, size_t file_size, uint32_t
912 int journalfile_v2_load(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
913 {
914 int ret, fd;
886 - uint64_t file_size;
887 - char path[RRDENG_PATH_MAX];
915 + char path_v1[RRDENG_PATH_MAX];
916 + char path_v2[RRDENG_PATH_MAX];
917 struct stat statbuf;
889 - uint32_t original_file_size = 0;
918 + size_t journal_v1_file_size = 0;
919 + size_t journal_v2_file_size;
920
891 - journalfile_generate_path(datafile, path, sizeof(path));
892 - ret = stat(path, &statbuf);
921 + journalfile_v1_generate_path(datafile, path_v1, sizeof(path_v1));
922 + ret = stat(path_v1, &statbuf);
923 if (!ret)
894 - original_file_size = (uint32_t)statbuf.st_size;
895 -
896 - journalfile_v2_generate_path(datafile, path, sizeof(path));
924 + journal_v1_file_size = (uint32_t)statbuf.st_size;
925
898 - fd = open(path, O_RDONLY);
926 + journalfile_v2_generate_path(datafile, path_v2, sizeof(path_v2));
927 + fd = open(path_v2, O_RDONLY);
928 if (fd < 0) {
929 if (errno == ENOENT)
930 return 1;
931 ++ctx->stats.fs_errors;
932 rrd_stat_atomic_add(&global_fs_errors, 1);
904 - error("DBENGINE: failed to open '%s'", path);
933 + error("DBENGINE: failed to open '%s'", path_v2);
934 return 1;
935 }
936
937 ret = fstat(fd, &statbuf);
938 if (ret) {
910 - error("DBENGINE: failed to get file information for '%s'", path);
939 + error("DBENGINE: failed to get file information for '%s'", path_v2);
940 close(fd);
941 return 1;
942 }
943
915 - file_size = (size_t)statbuf.st_size;
944 + journal_v2_file_size = (size_t)statbuf.st_size;
945
917 - if (file_size < sizeof(struct journal_v2_header)) {
918 - error_report("Invalid file %s. Not the expected size", path);
946 + if (journal_v2_file_size < sizeof(struct journal_v2_header)) {
947 + error_report("Invalid file %s. Not the expected size", path_v2);
948 close(fd);
949 return 1;
950 }
951
952 usec_t start_loading = now_realtime_usec();
924 - uint8_t *data_start = mmap(NULL, file_size, PROT_READ, MAP_SHARED, fd, 0);
953 + uint8_t *data_start = mmap(NULL, journal_v2_file_size, PROT_READ, MAP_SHARED, fd, 0);
954 if (data_start == MAP_FAILED) {
955 close(fd);
956 return 1;
957 }
958
930 - info("DBENGINE: checking integrity of '%s'", path);
931 - int rc = journalfile_v2_validate(data_start, file_size, original_file_size);
959 + info("DBENGINE: checking integrity of '%s'", path_v2);
960 + int rc = journalfile_v2_validate(data_start, journal_v2_file_size, journal_v1_file_size);
961 if (unlikely(rc)) {
962 if (rc == 2)
934 - error_report("File %s needs to be rebuilt", path);
963 + error_report("File %s needs to be rebuilt", path_v2);
964 else if (rc == 3)
936 - error_report("File %s will be skipped", path);
965 + error_report("File %s will be skipped", path_v2);
966 else
938 - error_report("File %s is invalid and it will be rebuilt", path);
967 + error_report("File %s is invalid and it will be rebuilt", path_v2);
968
940 - if (unlikely(munmap(data_start, file_size)))
941 - error("DBENGINE: failed to unmap '%s'", path);
969 + if (unlikely(munmap(data_start, journal_v2_file_size)))
970 + error("DBENGINE: failed to unmap '%s'", path_v2);
971
972 close(fd);
973 return rc;
@@ -948,15 +977,15 @@ int journalfile_v2_load(struct rrdengine_instance *ctx, struct rrdengine_journal
977 uint32_t entries = j2_header->metric_count;
978
979 if (unlikely(!entries)) {
951 - if (unlikely(munmap(data_start, file_size)))
952 - error("DBENGINE: failed to unmap '%s'", path);
980 + if (unlikely(munmap(data_start, journal_v2_file_size)))
981 + error("DBENGINE: failed to unmap '%s'", path_v2);
982
983 close(fd);
984 return 1;
985 }
986
958 - madvise_dontfork(data_start, file_size);
959 - madvise_dontdump(data_start, file_size);
987 + madvise_dontfork(data_start, journal_v2_file_size);
988 + madvise_dontdump(data_start, journal_v2_file_size);
989
990 struct journal_metric_list *metric = (struct journal_metric_list *) (data_start + j2_header->metric_offset);
991
@@ -978,11 +1007,11 @@ int journalfile_v2_load(struct rrdengine_instance *ctx, struct rrdengine_journal
1007 metric++;
1008 }
1009
981 - info("DBENGINE: journal file '%s' loaded (size:%"PRIu64") with %u metrics in %d ms", path, file_size, entries,
1010 + info("DBENGINE: journal file '%s' loaded (size:%"PRIu64") with %u metrics in %d ms", path_v2, journal_v2_file_size, entries,
1011 (int) ((now_realtime_usec() - start_loading) / USEC_PER_MS));
1012
1013 // Initialize the journal file to be able to access the data
985 - journalfile_v2_data_set(journalfile, fd, data_start, file_size);
1014 + journalfile_v2_data_set(journalfile, fd, data_start, journal_v2_file_size);
1015
1016 // File is OK load it
1017 return 0;
@@ -1025,7 +1054,7 @@ void *journalfile_v2_write_extent_list(Pvoid_t JudyL_extents_pos, void *data)
1054
1055 static int journalfile_verify_space(struct journal_v2_header *j2_header, void *data, uint32_t bytes)
1056 {
1028 - if ((unsigned long)(((uint8_t *) data - (uint8_t *) j2_header->data) + bytes) > (j2_header->total_file_size - sizeof(struct journal_v2_block_trailer)))
1057 + if ((unsigned long)(((uint8_t *) data - (uint8_t *) j2_header->data) + bytes) > (j2_header->journal_v2_file_size - sizeof(struct journal_v2_block_trailer)))
1058 return 1;
1059
1060 return 0;
@@ -1195,8 +1224,8 @@ void journalfile_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno
1224 j2_header.page_offset = pages_offset;
1225 j2_header.extent_trailer_offset = extent_offset_trailer;
1226 j2_header.metric_trailer_offset = metric_offset_trailer;
1198 - j2_header.total_file_size = total_file_size;
1199 - j2_header.original_file_size = (uint32_t) journalfile->pos;
1227 + j2_header.journal_v2_file_size = total_file_size;
1228 + j2_header.journal_v1_file_size = (uint32_t) journalfile->pos;
1229 j2_header.data = data_start; // Used during migration
1230
1231 struct journal_v2_block_trailer *journal_v2_trailer;
@@ -1360,7 +1389,7 @@ int journalfile_load(struct rrdengine_instance *ctx, struct rrdengine_journalfil
1389 }
1390 }
1391
1363 - journalfile_generate_path(datafile, path, sizeof(path));
1392 + journalfile_v1_generate_path(datafile, path, sizeof(path));
1393
1394 // If it is not the last file, open read only
1395 fd = open_file_direct_io(path, O_RDWR, &file);
database/engine/journalfile.h
+4 -4
@@ -38,7 +38,6 @@ struct rrdengine_journalfile {
38 int32_t refcount;
39 time_t first_time_s;
40 time_t last_time_s;
41 - size_t not_needed_counter;
41 time_t not_needed_since_s;
42 } v2;
43
@@ -115,8 +114,8 @@ struct journal_v2_header {
114 uint32_t page_offset;
115 uint32_t extent_trailer_offset; // CRC for entent list
116 uint32_t metric_trailer_offset; // CRC for metric list
118 - uint32_t original_file_size; // This is the original journal file
119 - uint32_t total_file_size; // This is the total file size
117 + uint32_t journal_v1_file_size; // This is the original journal file
118 + uint32_t journal_v2_file_size; // This is the total file size
119 void *data; // Used when building the index
120 };
121
@@ -131,7 +130,7 @@ struct transaction_commit_log {
130
131 struct wal;
132
134 -void journalfile_generate_path(struct rrdengine_datafile *datafile, char *str, size_t maxlen);
133 +void journalfile_v1_generate_path(struct rrdengine_datafile *datafile, char *str, size_t maxlen);
134 void journalfile_v2_generate_path(struct rrdengine_datafile *datafile, char *str, size_t maxlen);
135 struct rrdengine_journalfile *journalfile_alloc_and_init(struct rrdengine_datafile *datafile);
136 void wal_flush_transaction_buffer(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile, struct wal *wal, uv_loop_t *loop);
@@ -153,5 +152,6 @@ size_t journalfile_v2_data_size_get(struct rrdengine_journalfile *journalfile);
152 void journalfile_v2_data_set(struct rrdengine_journalfile *journalfile, int fd, void *journal_data, uint32_t journal_data_size);
153 struct journal_v2_header *journalfile_v2_data_acquire(struct rrdengine_journalfile *journalfile, size_t *data_size, time_t wanted_first_time_s, time_t wanted_last_time_s);
154 void journalfile_v2_data_release(struct rrdengine_journalfile *journalfile);
155 +void journalfile_v2_data_unmount_cleanup(time_t now_s);
156
157 #endif /* NETDATA_JOURNALFILE_H */
\ No newline at end of file
database/engine/pagecache.c
+8 -11
@@ -430,19 +430,16 @@ static size_t list_has_time_gaps(
430 pd->status &= ~PDC_PAGE_DISK_PENDING;
431 pd->status |= PDC_PAGE_READY | PDC_PAGE_PRELOADED | PDC_PAGE_PRELOADED_PASS4;
432 }
433 - else {
433 + else if(!(pd->status & PDC_PAGE_FAILED) && (pd->status & PDC_PAGE_DATAFILE_ACQUIRED)) {
434 (*pages_pending)++;
435
436 - if (pd->status & PDC_PAGE_DISK_PENDING) {
437 - internal_fatal(pd->status & PDC_PAGE_SKIP, "page is disk pending and skipped");
438 - internal_fatal(!pd->datafile.ptr, "datafile is NULL");
439 - internal_fatal(!pd->datafile.extent.bytes, "datafile.extent.bytes zero");
440 - internal_fatal(!pd->datafile.extent.pos, "datafile.extent.pos is zero");
441 - internal_fatal(!pd->datafile.fileno, "datafile.fileno is zero");
442 - }
443 - else
444 - internal_fatal(!(pd->status & PDC_PAGE_FAILED),
445 - "DBENGINE: pdc has a disk pending page, without proper tagging");
436 + pd->status |= PDC_PAGE_DISK_PENDING;
437 +
438 + internal_fatal(pd->status & PDC_PAGE_SKIP, "page is disk pending and skipped");
439 + internal_fatal(!pd->datafile.ptr, "datafile is NULL");
440 + internal_fatal(!pd->datafile.extent.bytes, "datafile.extent.bytes zero");
441 + internal_fatal(!pd->datafile.extent.pos, "datafile.extent.pos is zero");
442 + internal_fatal(!pd->datafile.fileno, "datafile.fileno is zero");
443 }
444 }
445 else {
database/engine/rrdengine.c
+22 -2
@@ -1242,7 +1242,7 @@ static void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_dat
1242 info("DBENGINE: deleting data and journal files to maintain disk quota");
1243 ret = journalfile_destroy_unsafe(journal_file, datafile);
1244 if (!ret) {
1245 - journalfile_generate_path(datafile, path, sizeof(path));
1245 + journalfile_v1_generate_path(datafile, path, sizeof(path));
1246 info("DBENGINE: deleted journal file \"%s\".", path);
1247 journalfile_v2_generate_path(datafile, path, sizeof(path));
1248 info("DBENGINE: deleted journal file \"%s\".", path);
@@ -1260,6 +1260,9 @@ static void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_dat
1260 ctx->disk_space -= deleted_bytes;
1261 info("DBENGINE: reclaimed %u bytes of disk space.", deleted_bytes);
1262
1263 + if (rrdeng_ctx_exceeded_disk_quota(ctx))
1264 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
1265 +
1266 rrdcontext_db_rotation();
1267 }
1268
@@ -1324,6 +1327,14 @@ unsigned rrdeng_target_data_file_size(struct rrdengine_instance *ctx) {
1327 return target_size;
1328 }
1329
1330 +bool rrdeng_ctx_exceeded_disk_quota(struct rrdengine_instance *ctx)
1331 +{
1332 + uint64_t estimated_disk_space = ctx->disk_space + rrdeng_target_data_file_size(ctx) -
1333 + (ctx->datafiles.first->prev ? ctx->datafiles.first->prev->pos : 0);
1334 +
1335 + return estimated_disk_space > ctx->max_disk_space;
1336 +}
1337 +
1338 /* return 0 on success */
1339 int init_rrd_files(struct rrdengine_instance *ctx)
1340 {
@@ -1471,6 +1482,15 @@ void timer_cb(uv_timer_t* handle) {
1482 epdl_cleanup1();
1483 deol_cleanup1();
1484
1485 + {
1486 + static time_t last_run_s = 0;
1487 + time_t now_s = now_monotonic_sec();
1488 + if(now_s - last_run_s >= 10) {
1489 + last_run_s = now_s;
1490 + journalfile_v2_data_unmount_cleanup(now_s);
1491 + }
1492 + }
1493 +
1494 #ifdef PDC_USE_JULYL
1495 julyl_cleanup1();
1496 #endif
@@ -1662,7 +1682,7 @@ void dbengine_event_loop(void* arg) {
1682 if (!ctx->worker_config.now_deleting_files &&
1683 ctx->datafiles.first->next != NULL &&
1684 ctx->datafiles.first->next->next != NULL &&
1665 - ctx->disk_space > MAX(ctx->max_disk_space, 2 * ctx->metric_API_max_producers * RRDENG_BLOCK_SIZE)) {
1685 + rrdeng_ctx_exceeded_disk_quota(ctx)) {
1686
1687 ctx->worker_config.now_deleting_files = true;
1688 if(!work_dispatch(ctx, NULL, NULL, opcode, database_rotate_tp_worker, after_database_rotate))
database/engine/rrdengine.h
+1 -1
@@ -378,7 +378,6 @@ struct rrdengine_instance {
378 int tier;
379 unsigned last_fileno; /* newest index of datafile and journalfile */
380 unsigned last_flush_fileno;
381 - unsigned long metric_API_max_producers;
381
382 bool create_new_datafile_pair;
383 uint8_t quiesce; /* set to SET_QUIESCE before shutdown of the engine */
@@ -398,6 +397,7 @@ void dbengine_page_free(void *page, size_t size);
397 void *dbengine_extent_alloc(size_t size);
398 void dbengine_extent_free(void *extent, size_t size);
399
400 +bool rrdeng_ctx_exceeded_disk_quota(struct rrdengine_instance *ctx);
401 int init_rrd_files(struct rrdengine_instance *ctx);
402 void finalize_rrd_files(struct rrdengine_instance *ctx);
403 bool rrdeng_dbengine_spawn(struct rrdengine_instance *ctx);
database/engine/rrdengineapi.c
-2
@@ -896,7 +896,6 @@ int rrdeng_init(RRDHOST *host, struct rrdengine_instance **ctxp, char *dbfiles_p
896 else
897 strncpyz(ctx->machine_guid, host->machine_guid, GUID_LEN);
898
899 - ctx->metric_API_max_producers = 0;
899 ctx->quiesce = NO_QUIESCE;
900 ctx->host = host;
901
@@ -1041,7 +1040,6 @@ RRDENG_SIZE_STATS rrdeng_size_statistics(struct rrdengine_instance *ctx) {
1040 uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
1041
1042 stats.currently_collected_metrics = ctx->stats.metric_API_producers;
1044 - stats.max_concurrently_collected_metrics = ctx->metric_API_max_producers;
1043
1044 internal_error(stats.metrics_pages != stats.extents_pages + stats.currently_collected_metrics,
1045 "DBENGINE: metrics pages is %zu, but extents pages is %zu and API consumers is %zu",
database/engine/rrdengineapi.h
-1
@@ -106,7 +106,6 @@ typedef struct rrdengine_size_statistics {
106 time_t last_time_s;
107
108 size_t currently_collected_metrics;
109 - size_t max_concurrently_collected_metrics;
109 size_t estimated_concurrently_collected_metrics;
110
111 size_t disk_space;
database/rrdhost.c
+4 -1
@@ -870,8 +870,11 @@ void dbengine_init(char *hostname) {
870 hostname, tier, dbenginepath);
871 break;
872 }
873 - else
873 + else {
874 + if (rrdeng_ctx_exceeded_disk_quota(multidb_ctx[created_tiers]))
875 + rrdeng_enq_cmd( multidb_ctx[created_tiers], RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
876 created_tiers++;
877 + }
878 }
879
880 if(created_tiers && created_tiers < storage_tiers) {
database/sqlite/sqlite_metadata.c
+1 -1
@@ -1372,7 +1372,7 @@ static void *metadata_unittest_threads(void)
1372 tu.join = 0;
1373 for (int i = 0; i < threads_to_create; i++) {
1374 char buf[100 + 1];
1375 - snprintf(buf, 100, "meta%d", i);
1375 + snprintf(buf, 100, "META[%d]", i);
1376 netdata_thread_create(
1377 &threads[i],
1378 buf,
libnetdata/log/log.c
+9 -1
@@ -986,7 +986,15 @@ void fatal_int( const char *file, const char *function, const unsigned long line
986 snprintfz(action_data, 70, "%04lu@%-10.10s:%-15.15s/%d", line, file, function, __errno);
987 char action_result[60+1];
988
989 - snprintfz(action_result, 60, "%s:%s", program_name, strncmp(thread_tag, "STREAM_RECEIVER", strlen("STREAM_RECEIVER")) ? thread_tag : "[x]");
989 + const char *tag_to_send = thread_tag;
990 +
991 + // anonymize thread names
992 + if(strncmp(thread_tag, THREAD_TAG_STREAM_RECEIVER, strlen(THREAD_TAG_STREAM_RECEIVER)) == 0)
993 + tag_to_send = THREAD_TAG_STREAM_RECEIVER;
994 + if(strncmp(thread_tag, THREAD_TAG_STREAM_SENDER, strlen(THREAD_TAG_STREAM_SENDER)) == 0)
995 + tag_to_send = THREAD_TAG_STREAM_SENDER;
996 +
997 + snprintfz(action_result, 60, "%s:%s", program_name, tag_to_send);
998 send_statistics("FATAL", action_result, action_data);
999
1000 #ifdef HAVE_BACKTRACE
ml/Host.cc
+2 -2
@@ -340,10 +340,10 @@ void Host::startAnomalyDetectionThreads() {
340
341 char Tag[NETDATA_THREAD_TAG_MAX + 1];
342
343 - snprintfz(Tag, NETDATA_THREAD_TAG_MAX, "TRAIN[%s]", rrdhost_hostname(RH));
343 + snprintfz(Tag, NETDATA_THREAD_TAG_MAX, "MLTR[%s]", rrdhost_hostname(RH));
344 netdata_thread_create(&TrainingThread, Tag, NETDATA_THREAD_OPTION_DEFAULT, train_main, static_cast<void *>(this));
345
346 - snprintfz(Tag, NETDATA_THREAD_TAG_MAX, "DETECT[%s]", rrdhost_hostname(RH));
346 + snprintfz(Tag, NETDATA_THREAD_TAG_MAX, "MLDT[%s]", rrdhost_hostname(RH));
347 netdata_thread_create(&DetectionThread, Tag, NETDATA_THREAD_OPTION_DEFAULT, detect_main, static_cast<void *>(this));
348 }
349
streaming/replication.c
+5 -3
@@ -1553,13 +1553,15 @@ void *replication_thread_main(void *ptr __maybe_unused) {
1553 threads = 1;
1554 }
1555
1556 - if(--threads) {
1556 + if(threads > 1) {
1557 replication_globals.main_thread.threads = threads;
1558 replication_globals.main_thread.threads_ptrs = mallocz(threads * sizeof(netdata_thread_t *));
1559
1560 - for(int i = 0; i < threads ;i++) {
1560 + for(int i = 1; i < threads ;i++) {
1561 + char tag[NETDATA_THREAD_TAG_MAX + 1];
1562 + snprintfz(tag, NETDATA_THREAD_TAG_MAX, "REPLAY[%d]", i + 1);
1563 replication_globals.main_thread.threads_ptrs[i] = mallocz(sizeof(netdata_thread_t));
1562 - netdata_thread_create(replication_globals.main_thread.threads_ptrs[i], "REPLICATION",
1564 + netdata_thread_create(replication_globals.main_thread.threads_ptrs[i], tag,
1565 NETDATA_THREAD_OPTION_JOINABLE, replication_worker_thread, NULL);
1566 }
1567 }
streaming/rrdpush.c
+2 -2
@@ -597,7 +597,7 @@ static void rrdpush_sender_thread_spawn(RRDHOST *host) {
597
598 if(!rrdhost_flag_check(host, RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN)) {
599 char tag[NETDATA_THREAD_TAG_MAX + 1];
600 - snprintfz(tag, NETDATA_THREAD_TAG_MAX, "STREAM_SENDER[%s]", rrdhost_hostname(host));
600 + snprintfz(tag, NETDATA_THREAD_TAG_MAX, THREAD_TAG_STREAM_SENDER "[%s]", rrdhost_hostname(host));
601
602 if(netdata_thread_create(&host->rrdpush_sender_thread, tag, NETDATA_THREAD_OPTION_DEFAULT, rrdpush_sender_thread, (void *) host->sender))
603 error("STREAM %s [send]: failed to create new thread for client.", rrdhost_hostname(host));
@@ -1036,7 +1036,7 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) {
1036 debug(D_SYSTEM, "starting STREAM receive thread.");
1037
1038 char tag[FILENAME_MAX + 1];
1039 - snprintfz(tag, FILENAME_MAX, "STREAM_RECEIVER[%s,[%s]:%s]", rpt->hostname, w->client_ip, w->client_port);
1039 + snprintfz(tag, FILENAME_MAX, THREAD_TAG_STREAM_RECEIVER "[%s,[%s]:%s]", rpt->hostname, w->client_ip, w->client_port);
1040
1041 if(netdata_thread_create(&rpt->thread, tag, NETDATA_THREAD_OPTION_DEFAULT, rrdpush_receiver_thread, (void *)rpt)) {
1042 rrdpush_receive_log_status(
streaming/rrdpush.h
+3
@@ -306,6 +306,9 @@ void *rrdpush_sender_thread(void *ptr);
306 void rrdpush_send_host_labels(RRDHOST *host);
307 void rrdpush_claimed_id(RRDHOST *host);
308
309 +#define THREAD_TAG_STREAM_RECEIVER "RCVR" // "[host]" is appended
310 +#define THREAD_TAG_STREAM_SENDER "SNDR" // "[host]" is appended
311 +
312 int rrdpush_receiver_thread_spawn(struct web_client *w, char *url);
313 void rrdpush_sender_thread_stop(RRDHOST *host, const char *reason, bool wait);
314
web/api/web_api_v1.c
-2
@@ -1544,7 +1544,6 @@ static void web_client_api_v1_dbengine_stats_for_tier(BUFFER *wb, size_t tier) {
1544 ",\n\t\t\"average_page_size_bytes\":%0.2f"
1545 ",\n\t\t\"estimated_concurrently_collected_metrics\":%zu"
1546 ",\n\t\t\"currently_collected_metrics\":%zu"
1547 - ",\n\t\t\"max_concurrently_collected_metrics\":%zu"
1547 ",\n\t\t\"disk_space\":%zu"
1548 ",\n\t\t\"max_disk_space\":%zu"
1549 , stats.default_granularity_secs
@@ -1573,7 +1572,6 @@ static void web_client_api_v1_dbengine_stats_for_tier(BUFFER *wb, size_t tier) {
1572 , stats.average_page_size_bytes
1573 , stats.estimated_concurrently_collected_metrics
1574 , stats.currently_collected_metrics
1576 - , stats.max_concurrently_collected_metrics
1575 , stats.disk_space
1576 , stats.max_disk_space
1577 );
web/server/static/static-threaded.c
+1 -1
@@ -539,7 +539,7 @@ void *socket_listen_main_static_threaded(void *ptr) {
539 static_workers_private_data[i].max_sockets = max_sockets / static_threaded_workers_count;
540
541 char tag[50 + 1];
542 - snprintfz(tag, 50, "WEB_SERVER[static%d]", i+1);
542 + snprintfz(tag, 50, "WEB[%d]", i+1);
543
544 info("starting worker %d", i+1);
545 netdata_thread_create(&static_workers_private_data[i].thread, tag, NETDATA_THREAD_OPTION_DEFAULT,