@cryptotaxi247 / netdata-1 / commits / a8ed7d2d0

Journal better estimations and watcher (#16467)

* get the number of lines per journal file * accurate estimates when using systemd 254+ * scan the journal files in reverse order, the newest first * disable running journalctl; use the file last modification timestamp if the messages cannot be retrieved * attempt to fix order of files scanned * find the timestamps of all boot_ids during header updates * added inotify watcher for detecting journal file changes * handle all cases in watcher * completed watcher * remove debug info * watcher now restarts when it gets inotify errors * removed log * do not remain the main thread of netdata

Costa Tsaousis committed Nov 24, 2023 at 16:48 UTC a8ed7d2d03de34a5260c7691a8d48f1fd7a5fe5e
14 files changed +981 -174
Makefile.am
+1
@@ -330,6 +330,7 @@ SYSTEMD_JOURNAL_PLUGIN_FILES = \
330 collectors/systemd-journal.plugin/systemd-main.c \
331 collectors/systemd-journal.plugin/systemd-units.c \
332 collectors/systemd-journal.plugin/systemd-journal.c \
333 + collectors/systemd-journal.plugin/systemd-journal-watcher.c \
334 collectors/systemd-journal.plugin/systemd-journal-annotations.c \
335 collectors/systemd-journal.plugin/systemd-journal-files.c \
336 collectors/systemd-journal.plugin/systemd-journal-fstat.c \
collectors/proc.plugin/sys_fs_btrfs.c
+4 -4
@@ -392,14 +392,14 @@ static inline int find_btrfs_devices(BTRFS_NODE *node, const char *path) {
392 continue;
393 }
394
395 - collector_info("BTRFS: device found '%s'", de->d_name);
395 + // internal_error("BTRFS: device found '%s'", de->d_name);
396
397 // --------------------------------------------------------------------
398 // search for it
399
400 for(d = node->devices ; d ; d = d->next) {
401 if(str2ll(de->d_name, NULL) == d->id){
402 - collector_info("BTRFS: existing device id '%d'", d->id);
402 + // collector_info("BTRFS: existing device id '%d'", d->id);
403 break;
404 }
405 }
@@ -411,11 +411,11 @@ static inline int find_btrfs_devices(BTRFS_NODE *node, const char *path) {
411 d = callocz(sizeof(BTRFS_DEVICE), 1);
412
413 d->id = str2ll(de->d_name, NULL);
414 - collector_info("BTRFS: new device with id '%d'", d->id);
414 + // collector_info("BTRFS: new device with id '%d'", d->id);
415
416 snprintfz(filename, FILENAME_MAX, "%s/%d/error_stats", path, d->id);
417 d->error_stats_filename = strdupz(filename);
418 - collector_info("BTRFS: error_stats_filename '%s'", filename);
418 + // collector_info("BTRFS: error_stats_filename '%s'", filename);
419
420 // link it
421 d->next = node->devices;
collectors/systemd-journal.plugin/systemd-internals.h
+46 -1
@@ -18,6 +18,9 @@
18 #define SYSTEMD_UNITS_FUNCTION_NAME "systemd-list-units"
19 #define SYSTEMD_UNITS_DEFAULT_TIMEOUT 30
20
21 +#define EXECUTE_WATCHER_PENDING_EVERY_MS 500
22 +#define FULL_JOURNAL_SCAN_EVERY_USEC (5 * 60 * USEC_PER_SEC)
23 +
24 extern __thread size_t fstat_thread_calls;
25 extern __thread size_t fstat_thread_cached_responses;
26 void fstat_cache_enable_on_thread(void);
@@ -55,10 +58,20 @@ struct journal_file {
58 usec_t file_last_modified_ut;
59 usec_t msg_first_ut;
60 usec_t msg_last_ut;
58 - usec_t last_scan_ut;
61 size_t size;
62 bool logged_failure;
63 + bool logged_journalctl_failure;
64 usec_t max_journal_vs_realtime_delta_ut;
65 +
66 + usec_t last_scan_monotonic_ut;
67 + usec_t last_scan_header_vs_last_modified_ut;
68 +
69 + uint64_t first_seqnum;
70 + uint64_t last_seqnum;
71 + sd_id128_t first_writer_id;
72 + sd_id128_t last_writer_id;
73 +
74 + uint64_t messages_in_file;
75 };
76
77 #define SDJF_SOURCE_ALL_NAME "all"
@@ -84,6 +97,7 @@ int journal_file_dict_items_forward_compar(const void *a, const void *b);
97 void buffer_json_journal_versions(BUFFER *wb);
98 void available_journal_file_sources_to_json_array(BUFFER *wb);
99 bool journal_files_completed_once(void);
100 +void journal_files_updater_all_headers_sorted(void);
101
102 FACET_ROW_SEVERITY syslog_priority_to_facet_severity(FACETS *facets, FACET_ROW *row, void *data);
103
@@ -97,14 +111,26 @@ void netdata_systemd_journal_transform_gid(FACETS *facets, BUFFER *wb, FACETS_TR
111 void netdata_systemd_journal_transform_cap_effective(FACETS *facets, BUFFER *wb, FACETS_TRANSFORMATION_SCOPE scope, void *data);
112 void netdata_systemd_journal_transform_timestamp_usec(FACETS *facets, BUFFER *wb, FACETS_TRANSFORMATION_SCOPE scope, void *data);
113
114 +usec_t journal_file_update_annotation_boot_id(sd_journal *j, struct journal_file *jf, const char *boot_id);
115 +
116 +#define MAX_JOURNAL_DIRECTORIES 100
117 +struct journal_directory {
118 + char *path;
119 + bool logged_failure;
120 +};
121 +extern struct journal_directory journal_directories[MAX_JOURNAL_DIRECTORIES];
122 +
123 void journal_init_files_and_directories(void);
124 void journal_init_query_status(void);
125 void function_systemd_journal(const char *transaction, char *function, int timeout, bool *cancelled);
126 void journal_files_registry_update(void);
127 +void journal_file_update_header(const char *filename, struct journal_file *jf);
128
129 void netdata_systemd_journal_message_ids_init(void);
130 void netdata_systemd_journal_transform_message_id(FACETS *facets __maybe_unused, BUFFER *wb, FACETS_TRANSFORMATION_SCOPE scope __maybe_unused, void *data __maybe_unused);
131
132 +void *journal_watcher_main(void *arg);
133 +
134 #ifdef ENABLE_SYSTEMD_DBUS
135 void function_systemd_units(const char *transaction, char *function, int timeout, bool *cancelled);
136 #endif
@@ -116,4 +142,23 @@ static inline void send_newline_and_flush(void) {
142 netdata_mutex_unlock(&stdout_mutex);
143 }
144
145 +static inline bool parse_journal_field(const char *data, size_t data_length, const char **key, size_t *key_length, const char **value, size_t *value_length) {
146 + const char *k = data;
147 + const char *equal = strchr(k, '=');
148 + if(unlikely(!equal))
149 + return false;
150 +
151 + size_t kl = equal - k;
152 +
153 + const char *v = ++equal;
154 + size_t vl = data_length - kl - 1;
155 +
156 + *key = k;
157 + *key_length = kl;
158 + *value = v;
159 + *value_length = vl;
160 +
161 + return true;
162 +}
163 +
164 #endif //NETDATA_COLLECTORS_SYSTEMD_INTERNALS_H
collectors/systemd-journal.plugin/systemd-journal-annotations.c
+4 -46
@@ -400,6 +400,7 @@ void netdata_systemd_journal_transform_boot_id(FACETS *facets __maybe_unused, BU
400 usec_t ut = UINT64_MAX;
401 usec_t *p_ut = dictionary_get(boot_ids_to_first_ut, boot_id);
402 if(!p_ut) {
403 +#ifndef HAVE_SD_JOURNAL_RESTART_FIELDS
404 struct journal_file *jf;
405 dfe_start_read(journal_files_registry, jf) {
406 const char *files[2] = {
@@ -416,59 +417,16 @@ void netdata_systemd_journal_transform_boot_id(FACETS *facets __maybe_unused, BU
417 continue;
418 }
419
419 - char m[100];
420 - size_t len = snprintfz(m, sizeof(m), "_BOOT_ID=%s", boot_id);
421 -
422 - r = sd_journal_add_match(j, m, len);
423 - if(r < 0) {
424 - internal_error(true, "JOURNAL: while looking for the first timestamp of boot_id '%s', "
425 - "sd_journal_add_match('%s') on file '%s' returned %d",
426 - boot_id, m, jf_dfe.name, r);
427 - sd_journal_close(j);
428 - continue;
429 - }
430 -
431 - r = sd_journal_seek_head(j);
432 - if(r < 0) {
433 - internal_error(true, "JOURNAL: while looking for the first timestamp of boot_id '%s', "
434 - "sd_journal_seek_head() on file '%s' returned %d",
435 - boot_id, jf_dfe.name, r);
436 - sd_journal_close(j);
437 - continue;
438 - }
439 -
440 - r = sd_journal_next(j);
441 - if(r < 0) {
442 - internal_error(true, "JOURNAL: while looking for the first timestamp of boot_id '%s', "
443 - "sd_journal_next() on file '%s' returned %d",
444 - boot_id, jf_dfe.name, r);
445 - sd_journal_close(j);
446 - continue;
447 - }
448 -
449 - usec_t t_ut = 0;
450 - r = sd_journal_get_realtime_usec(j, &t_ut);
451 - if(r < 0 || !t_ut) {
452 - internal_error(r != -EADDRNOTAVAIL, "JOURNAL: while looking for the first timestamp of boot_id '%s', "
453 - "sd_journal_get_realtime_usec() on file '%s' returned %d",
454 - boot_id, jf_dfe.name, r);
455 - sd_journal_close(j);
456 - continue;
457 - }
458 -
459 - if(t_ut < ut)
460 - ut = t_ut;
461 -
420 + ut = journal_file_update_annotation_boot_id(j, jf, boot_id);
421 sd_journal_close(j);
422 }
423 dfe_done(jf);
465 -
466 - dictionary_set(boot_ids_to_first_ut, boot_id, &ut, sizeof(ut));
424 +#endif
425 }
426 else
427 ut = *p_ut;
428
471 - if(ut != UINT64_MAX) {
429 + if(ut && ut != UINT64_MAX) {
430 char buffer[ISO8601_MAX_LENGTH];
431 iso8601_datetime_ut(buffer, sizeof(buffer), ut, ISO8601_UTC);
432
collectors/systemd-journal.plugin/systemd-journal-files.c
+371 -39
@@ -4,14 +4,8 @@
4
5 #define SYSTEMD_JOURNAL_MAX_SOURCE_LEN 64
6 #define VAR_LOG_JOURNAL_MAX_DEPTH 10
7 -#define MAX_JOURNAL_DIRECTORIES 100
7
9 -struct journal_directory {
10 - char *path;
11 - bool logged_failure;
12 -};
13 -
14 -static struct journal_directory journal_directories[MAX_JOURNAL_DIRECTORIES] = { 0 };
8 +struct journal_directory journal_directories[MAX_JOURNAL_DIRECTORIES] = { 0 };
9 DICTIONARY *journal_files_registry = NULL;
10 DICTIONARY *used_hashes_registry = NULL;
11
@@ -26,7 +20,200 @@ void buffer_json_journal_versions(BUFFER *wb) {
20 buffer_json_object_close(wb);
21 }
22
29 -static void journal_file_update_msg_ut(const char *filename, struct journal_file *jf) {
23 +static bool journal_sd_id128_parse(const char *in, sd_id128_t *ret) {
24 + while(isspace(*in))
25 + in++;
26 +
27 + char uuid[33];
28 + strncpyz(uuid, in, 32);
29 + uuid[32] = '\0';
30 +
31 + if(strlen(uuid) == 32) {
32 + sd_id128_t read;
33 + if(sd_id128_from_string(uuid, &read) == 0) {
34 + *ret = read;
35 + return true;
36 + }
37 + }
38 +
39 + return false;
40 +}
41 +
42 +static void journal_file_get_header_from_journalctl(const char *filename, struct journal_file *jf) {
43 + // unfortunately, our capabilities are not inheritted by journalctl
44 + // so, it fails to give us the information we need.
45 +
46 + bool read_writer = false, read_head = false, read_tail = false;
47 +
48 + char cmd[FILENAME_MAX * 2];
49 + snprintfz(cmd, sizeof(cmd), "journalctl --header --file '%s'", filename);
50 + CLEAN_BUFFER *wb = run_command_and_get_output_to_buffer(cmd, 1024);
51 + if(wb) {
52 + const char *s = buffer_tostring(wb);
53 +
54 + const char *sequential_id_header = "Sequential Number ID:";
55 + const char *sequential_id_data = strcasestr(s, sequential_id_header);
56 + if(sequential_id_data) {
57 + sequential_id_data += strlen(sequential_id_header);
58 + if(journal_sd_id128_parse(sequential_id_data, &jf->first_writer_id))
59 + read_writer = true;
60 + }
61 +
62 + const char *head_sequential_number_header = "Head sequential number:";
63 + const char *head_sequential_number_data = strcasestr(s, head_sequential_number_header);
64 + if(head_sequential_number_data) {
65 + head_sequential_number_data += strlen(head_sequential_number_header);
66 +
67 + while(isspace(*head_sequential_number_data))
68 + head_sequential_number_data++;
69 +
70 + if(isdigit(*head_sequential_number_data)) {
71 + jf->first_seqnum = strtoul(head_sequential_number_data, NULL, 10);
72 + if(jf->first_seqnum)
73 + read_head = true;
74 + }
75 + }
76 +
77 + const char *tail_sequential_number_header = "Tail sequential number:";
78 + const char *tail_sequential_number_data = strcasestr(s, tail_sequential_number_header);
79 + if(tail_sequential_number_data) {
80 + tail_sequential_number_data += strlen(tail_sequential_number_header);
81 +
82 + while(isspace(*tail_sequential_number_data))
83 + tail_sequential_number_data++;
84 +
85 + if(isdigit(*tail_sequential_number_data)) {
86 + jf->last_seqnum = strtoul(tail_sequential_number_data, NULL, 10);
87 + if(jf->last_seqnum)
88 + read_tail = true;
89 + }
90 + }
91 +
92 + if(read_head && read_tail && jf->last_seqnum > jf->first_seqnum)
93 + jf->messages_in_file = jf->last_seqnum - jf->first_seqnum;
94 + }
95 +
96 + if(!jf->logged_journalctl_failure && (!read_head || !read_head || !read_tail)) {
97 +
98 + nd_log(NDLS_COLLECTORS, NDLP_NOTICE,
99 + "Failed to read %s%s%s from journalctl's output on filename '%s', using the command: %s",
100 + read_writer?"":"writer id,",
101 + read_head?"":"head id,",
102 + read_tail?"":"tail id,",
103 + filename, cmd);
104 +
105 + jf->logged_journalctl_failure = true;
106 + }
107 +}
108 +
109 +usec_t journal_file_update_annotation_boot_id(sd_journal *j, struct journal_file *jf, const char *boot_id) {
110 + usec_t ut = UINT64_MAX;
111 + int r;
112 +
113 + char m[100];
114 + size_t len = snprintfz(m, sizeof(m), "_BOOT_ID=%s", boot_id);
115 +
116 + sd_journal_flush_matches(j);
117 +
118 + r = sd_journal_add_match(j, m, len);
119 + if(r < 0) {
120 + errno = -r;
121 + internal_error(true,
122 + "JOURNAL: while looking for the first timestamp of boot_id '%s', "
123 + "sd_journal_add_match('%s') on file '%s' returned %d",
124 + boot_id, m, jf->filename, r);
125 + return UINT64_MAX;
126 + }
127 +
128 + r = sd_journal_seek_head(j);
129 + if(r < 0) {
130 + errno = -r;
131 + internal_error(true,
132 + "JOURNAL: while looking for the first timestamp of boot_id '%s', "
133 + "sd_journal_seek_head() on file '%s' returned %d",
134 + boot_id, jf->filename, r);
135 + return UINT64_MAX;
136 + }
137 +
138 + r = sd_journal_next(j);
139 + if(r < 0) {
140 + errno = -r;
141 + internal_error(true,
142 + "JOURNAL: while looking for the first timestamp of boot_id '%s', "
143 + "sd_journal_next() on file '%s' returned %d",
144 + boot_id, jf->filename, r);
145 + return UINT64_MAX;
146 + }
147 +
148 + r = sd_journal_get_realtime_usec(j, &ut);
149 + if(r < 0 || !ut || ut == UINT64_MAX) {
150 + errno = -r;
151 + internal_error(r != -EADDRNOTAVAIL,
152 + "JOURNAL: while looking for the first timestamp of boot_id '%s', "
153 + "sd_journal_get_realtime_usec() on file '%s' returned %d",
154 + boot_id, jf->filename, r);
155 + return UINT64_MAX;
156 + }
157 +
158 + if(ut && ut != UINT64_MAX) {
159 + dictionary_set(boot_ids_to_first_ut, boot_id, &ut, sizeof(ut));
160 + return ut;
161 + }
162 +
163 + return UINT64_MAX;
164 +}
165 +
166 +static void journal_file_get_boot_id_annotations(sd_journal *j __maybe_unused, struct journal_file *jf __maybe_unused) {
167 +#ifdef HAVE_SD_JOURNAL_RESTART_FIELDS
168 + sd_journal_flush_matches(j);
169 +
170 + int r = sd_journal_query_unique(j, "_BOOT_ID");
171 + if (r < 0) {
172 + errno = -r;
173 + internal_error(true,
174 + "JOURNAL: while querying for the unique _BOOT_ID values, "
175 + "sd_journal_query_unique() on file '%s' returned %d",
176 + jf->filename, r);
177 + errno = -r;
178 + return;
179 + }
180 +
181 + const void *data = NULL;
182 + size_t data_length;
183 +
184 + DICTIONARY *dict = dictionary_create(DICT_OPTION_SINGLE_THREADED);
185 +
186 + SD_JOURNAL_FOREACH_UNIQUE(j, data, data_length) {
187 + const char *key, *value;
188 + size_t key_length, value_length;
189 +
190 + if(!parse_journal_field(data, data_length, &key, &key_length, &value, &value_length))
191 + continue;
192 +
193 + if(value_length != 32)
194 + continue;
195 +
196 + char buf[33];
197 + memcpy(buf, value, 32);
198 + buf[32] = '\0';
199 +
200 + dictionary_set(dict, buf, NULL, 0);
201 + }
202 +
203 + void *nothing;
204 + dfe_start_read(dict, nothing){
205 + journal_file_update_annotation_boot_id(j, jf, nothing_dfe.name);
206 + }
207 + dfe_done(nothing);
208 +
209 + dictionary_destroy(dict);
210 +#endif
211 +}
212 +
213 +void journal_file_update_header(const char *filename, struct journal_file *jf) {
214 + if(jf->last_scan_header_vs_last_modified_ut == jf->file_last_modified_ut)
215 + return;
216 +
217 fstat_cache_enable_on_thread();
218
219 const char *files[2] = {
@@ -46,23 +233,41 @@ static void journal_file_update_msg_ut(const char *filename, struct journal_file
233
234 jf->msg_first_ut = 0;
235 jf->msg_last_ut = jf->file_last_modified_ut;
236 + jf->last_scan_header_vs_last_modified_ut = jf->file_last_modified_ut;
237 return;
238 }
239
240 usec_t first_ut = 0, last_ut = 0;
241 + uint64_t first_seqnum = 0, last_seqnum = 0;
242 + sd_id128_t first_writer_id = SD_ID128_NULL, last_writer_id = SD_ID128_NULL;
243
244 if(sd_journal_seek_head(j) < 0 || sd_journal_next(j) < 0 || sd_journal_get_realtime_usec(j, &first_ut) < 0 || !first_ut) {
245 internal_error(true, "cannot find the timestamp of the first message in '%s'", filename);
246 first_ut = 0;
247 }
248 +#ifdef HAVE_SD_JOURNAL_GET_SEQNUM
249 + else {
250 + if(sd_journal_get_seqnum(j, &first_seqnum, &first_writer_id) < 0 || !first_seqnum) {
251 + internal_error(true, "cannot find the first seqnums of the first message in '%s'", filename);
252 + first_seqnum = 0;
253 + memset(&first_writer_id, 0, sizeof(first_writer_id));
254 + }
255 + }
256 +#endif
257
258 if(sd_journal_seek_tail(j) < 0 || sd_journal_previous(j) < 0 || sd_journal_get_realtime_usec(j, &last_ut) < 0 || !last_ut) {
259 internal_error(true, "cannot find the timestamp of the last message in '%s'", filename);
260 last_ut = jf->file_last_modified_ut;
261 }
63 -
64 - sd_journal_close(j);
65 - fstat_cache_disable_on_thread();
262 +#ifdef HAVE_SD_JOURNAL_GET_SEQNUM
263 + else {
264 + if(sd_journal_get_seqnum(j, &last_seqnum, &last_writer_id) < 0 || !last_seqnum) {
265 + internal_error(true, "cannot find the last seqnums of the first message in '%s'", filename);
266 + last_seqnum = 0;
267 + memset(&last_writer_id, 0, sizeof(last_writer_id));
268 + }
269 + }
270 +#endif
271
272 if(first_ut > last_ut) {
273 internal_error(true, "timestamps are flipped in file '%s'", filename);
@@ -71,8 +276,77 @@ static void journal_file_update_msg_ut(const char *filename, struct journal_file
276 last_ut = t;
277 }
278
279 + if(!first_seqnum || !first_ut) {
280 + // extract these from the filename - if possible
281 +
282 + const char *at = strchr(filename, '@');
283 + if(at) {
284 + const char *dash_seqnum = strchr(at + 1, '-');
285 + if(dash_seqnum) {
286 + const char *dash_first_msg_ut = strchr(dash_seqnum + 1, '-');
287 + if(dash_first_msg_ut) {
288 + const char *dot_journal = strstr(dash_first_msg_ut + 1, ".journal");
289 + if(dot_journal) {
290 + if(dash_seqnum - at - 1 == 32 &&
291 + dash_first_msg_ut - dash_seqnum - 1 == 16 &&
292 + dot_journal - dash_first_msg_ut - 1 == 16) {
293 + sd_id128_t writer;
294 + if(journal_sd_id128_parse(at + 1, &writer)) {
295 + char *endptr = NULL;
296 + uint64_t seqnum = strtoul(dash_seqnum + 1, &endptr, 16);
297 + if(endptr == dash_first_msg_ut) {
298 + uint64_t ts = strtoul(dash_first_msg_ut + 1, &endptr, 16);
299 + if(endptr == dot_journal) {
300 + first_seqnum = seqnum;
301 + first_writer_id = writer;
302 + first_ut = ts;
303 + }
304 + }
305 + }
306 + }
307 + }
308 + }
309 + }
310 + }
311 + }
312 +
313 + jf->first_seqnum = first_seqnum;
314 + jf->last_seqnum = last_seqnum;
315 +
316 + jf->first_writer_id = first_writer_id;
317 + jf->last_writer_id = last_writer_id;
318 +
319 jf->msg_first_ut = first_ut;
320 jf->msg_last_ut = last_ut;
321 +
322 + if(!jf->msg_last_ut)
323 + jf->msg_last_ut = jf->file_last_modified_ut;
324 +
325 + if(last_seqnum > first_seqnum) {
326 + if(!sd_id128_equal(first_writer_id, last_writer_id)) {
327 + jf->messages_in_file = 0;
328 + nd_log(NDLS_COLLECTORS, NDLP_NOTICE,
329 + "The writers of the first and the last message in file '%s' differ."
330 + , filename);
331 + }
332 + else
333 + jf->messages_in_file = last_seqnum - first_seqnum + 1;
334 + }
335 + else
336 + jf->messages_in_file = 0;
337 +
338 +// if(!jf->messages_in_file)
339 +// journal_file_get_header_from_journalctl(filename, jf);
340 +
341 + journal_file_get_boot_id_annotations(j, jf);
342 + sd_journal_close(j);
343 + fstat_cache_disable_on_thread();
344 +
345 + jf->last_scan_header_vs_last_modified_ut = jf->file_last_modified_ut;
346 +
347 + nd_log(NDLS_COLLECTORS, NDLP_DEBUG,
348 + "Journal file header updated '%s'",
349 + jf->filename);
350 }
351
352 static STRING *string_strdupz_source(const char *s, const char *e, size_t max_len, const char *prefix) {
@@ -166,38 +440,29 @@ static void files_registry_insert_cb(const DICTIONARY_ITEM *item, void *value, v
440 else
441 jf->source_type |= SDJF_LOCAL_ALL | SDJF_LOCAL_OTHER;
442
169 - journal_file_update_msg_ut(jf->filename, jf);
443 + jf->msg_last_ut = jf->file_last_modified_ut;
444
171 - internal_error(true,
172 - "found journal file '%s', type %d, source '%s', "
173 - "file modified: %"PRIu64", "
174 - "msg {first: %"PRIu64", last: %"PRIu64"}",
175 - jf->filename, jf->source_type, jf->source ? string2str(jf->source) : "<unset>",
176 - jf->file_last_modified_ut,
177 - jf->msg_first_ut, jf->msg_last_ut);
445 + nd_log(NDLS_COLLECTORS, NDLP_DEBUG,
446 + "Journal file added to the journal files registry: '%s'",
447 + jf->filename);
448 }
449
450 static bool files_registry_conflict_cb(const DICTIONARY_ITEM *item, void *old_value, void *new_value, void *data __maybe_unused) {
451 struct journal_file *jf = old_value;
452 struct journal_file *njf = new_value;
453
184 - if(njf->last_scan_ut > jf->last_scan_ut)
185 - jf->last_scan_ut = njf->last_scan_ut;
454 + if(njf->last_scan_monotonic_ut > jf->last_scan_monotonic_ut)
455 + jf->last_scan_monotonic_ut = njf->last_scan_monotonic_ut;
456
457 if(njf->file_last_modified_ut > jf->file_last_modified_ut) {
458 jf->file_last_modified_ut = njf->file_last_modified_ut;
459 jf->size = njf->size;
460
191 - const char *filename = dictionary_acquired_item_name(item);
192 - journal_file_update_msg_ut(filename, jf);
461 + jf->msg_last_ut = jf->file_last_modified_ut;
462
194 -// internal_error(true,
195 -// "updated journal file '%s', type %d, "
196 -// "file modified: %"PRIu64", "
197 -// "msg {first: %"PRIu64", last: %"PRIu64"}",
198 -// filename, jf->source_type,
199 -// jf->file_last_modified_ut,
200 -// jf->msg_first_ut, jf->msg_last_ut);
463 + nd_log(NDLS_COLLECTORS, NDLP_DEBUG,
464 + "Journal file updated to the journal files registry '%s'",
465 + jf->filename);
466 }
467
468 return false;
@@ -337,7 +602,7 @@ static void files_registry_delete_cb(const DICTIONARY_ITEM *item, void *value, v
602 string_freez(jf->source);
603 }
604
340 -void journal_directory_scan(const char *dirname, int depth, usec_t last_scan_ut) {
605 +void journal_directory_scan(const char *dirname, int depth, usec_t last_scan_monotonic_ut) {
606 static const char *ext = ".journal";
607 static const size_t ext_len = sizeof(".journal") - 1;
608
@@ -367,7 +632,7 @@ void journal_directory_scan(const char *dirname, int depth, usec_t last_scan_ut)
632 if (S_ISDIR(info.st_mode)) {
633 // If entry is a directory, call traverse recursively.
634 if (strcmp(entry->d_name, ".") != 0 && strcmp(entry->d_name, "..") != 0)
370 - journal_directory_scan(absolute_path, depth + 1, last_scan_ut);
635 + journal_directory_scan(absolute_path, depth + 1, last_scan_monotonic_ut);
636
637 }
638 else if (S_ISREG(info.st_mode)) {
@@ -378,7 +643,7 @@ void journal_directory_scan(const char *dirname, int depth, usec_t last_scan_ut)
643 if (len > ext_len && strcmp(filename + len - ext_len, ext) == 0) {
644 struct journal_file t = {
645 .file_last_modified_ut = info.st_mtim.tv_sec * USEC_PER_SEC + info.st_mtim.tv_nsec / NSEC_PER_USEC,
381 - .last_scan_ut = last_scan_ut,
646 + .last_scan_monotonic_ut = last_scan_monotonic_ut,
647 .size = info.st_size,
648 .max_journal_vs_realtime_delta_ut = JOURNAL_VS_REALTIME_DELTA_DEFAULT_UT,
649 };
@@ -396,28 +661,71 @@ bool journal_files_completed_once(void) {
661 return journal_files_scans > 0;
662 }
663
664 +static int journal_file_dict_items_last_modified_compar(const void *a, const void *b) {
665 + const DICTIONARY_ITEM **ad = (const DICTIONARY_ITEM **)a, **bd = (const DICTIONARY_ITEM **)b;
666 + struct journal_file *jfa = dictionary_acquired_item_value(*ad);
667 + struct journal_file *jfb = dictionary_acquired_item_value(*bd);
668 +
669 + if(jfa->file_last_modified_ut > jfb->file_last_modified_ut)
670 + return -1;
671 + else if(jfa->file_last_modified_ut < jfb->file_last_modified_ut)
672 + return 1;
673 +
674 + return 0;
675 +}
676 +
677 +void journal_files_updater_all_headers_sorted(void) {
678 + const DICTIONARY_ITEM *file_items[dictionary_entries(journal_files_registry)];
679 + size_t files_used = 0;
680 +
681 + struct journal_file *jf;
682 + dfe_start_write(journal_files_registry, jf){
683 + if(jf->last_scan_header_vs_last_modified_ut < jf->file_last_modified_ut)
684 + file_items[files_used++] = dictionary_acquired_item_dup(journal_files_registry, jf_dfe.item);
685 + }
686 + dfe_done(jf);
687 +
688 + // sort them in reverse order (newer first)
689 + qsort(file_items, files_used, sizeof(const DICTIONARY_ITEM *),
690 + journal_file_dict_items_last_modified_compar);
691 +
692 + // update the header (first and last message ut, sequence numbers, etc)
693 + for(size_t i = 0; i < files_used ; i++) {
694 + jf = dictionary_acquired_item_value(file_items[i]);
695 + journal_file_update_header(jf->filename, jf);
696 + dictionary_acquired_item_release(journal_files_registry, file_items[i]);
697 + send_newline_and_flush();
698 + }
699 +}
700 +
701 void journal_files_registry_update(void) {
702 static SPINLOCK spinlock = NETDATA_SPINLOCK_INITIALIZER;
703
704 if(spinlock_trylock(&spinlock)) {
403 - usec_t scan_ut = now_monotonic_usec();
705 + usec_t scan_monotonic_ut = now_monotonic_usec();
706
707 for(unsigned i = 0; i < MAX_JOURNAL_DIRECTORIES; i++) {
708 if(!journal_directories[i].path)
709 break;
710
409 - journal_directory_scan(journal_directories[i].path, 0, scan_ut);
711 + journal_directory_scan(journal_directories[i].path, 0, scan_monotonic_ut);
712 }
713
714 struct journal_file *jf;
715 dfe_start_write(journal_files_registry, jf){
414 - if(jf->last_scan_ut < scan_ut)
415 - dictionary_del(journal_files_registry, jf_dfe.name);
416 - }
716 + if(jf->last_scan_monotonic_ut < scan_monotonic_ut)
717 + dictionary_del(journal_files_registry, jf_dfe.name);
718 + }
719 dfe_done(jf);
720
721 + journal_files_updater_all_headers_sorted();
722 +
723 journal_files_scans++;
724 spinlock_unlock(&spinlock);
725 +
726 + internal_error(true,
727 + "Journal library scan completed in %.3f ms",
728 + (double)(now_monotonic_usec() - scan_monotonic_ut) / (double)USEC_PER_MS);
729 }
730 }
731
@@ -428,12 +736,21 @@ int journal_file_dict_items_backward_compar(const void *a, const void *b) {
736 struct journal_file *jfa = dictionary_acquired_item_value(*ad);
737 struct journal_file *jfb = dictionary_acquired_item_value(*bd);
738
739 + // compare the last message timestamps
740 if(jfa->msg_last_ut < jfb->msg_last_ut)
741 return 1;
742
743 if(jfa->msg_last_ut > jfb->msg_last_ut)
744 return -1;
745
746 + // compare the file last modification timestamps
747 + if(jfa->file_last_modified_ut < jfb->file_last_modified_ut)
748 + return 1;
749 +
750 + if(jfa->file_last_modified_ut > jfb->file_last_modified_ut)
751 + return -1;
752 +
753 + // compare the first message timestamps
754 if(jfa->msg_first_ut < jfb->msg_first_ut)
755 return 1;
756
@@ -447,14 +764,26 @@ int journal_file_dict_items_forward_compar(const void *a, const void *b) {
764 return -journal_file_dict_items_backward_compar(a, b);
765 }
766
767 +static bool boot_id_conflict_cb(const DICTIONARY_ITEM *item, void *old_value, void *new_value, void *data __maybe_unused) {
768 + usec_t *old_usec = old_value;
769 + usec_t *new_usec = new_value;
770 +
771 + if(*new_usec < *old_usec) {
772 + *old_usec = *new_usec;
773 + return true;
774 + }
775 +
776 + return false;
777 +}
778 +
779 void journal_init_files_and_directories(void) {
780 unsigned d = 0;
781
782 // ------------------------------------------------------------------------
783 // setup the journal directories
784
456 - journal_directories[d++].path = strdupz("/var/log/journal");
785 journal_directories[d++].path = strdupz("/run/log/journal");
786 + journal_directories[d++].path = strdupz("/var/log/journal");
787
788 if(*netdata_configured_host_prefix) {
789 char path[PATH_MAX];
@@ -485,4 +814,7 @@ void journal_init_files_and_directories(void) {
814 boot_ids_to_first_ut = dictionary_create_advanced(
815 DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_FIXED_SIZE,
816 NULL, sizeof(usec_t));
817 +
818 + dictionary_register_conflict_callback(boot_ids_to_first_ut, boot_id_conflict_cb, NULL);
819 +
820 }
collectors/systemd-journal.plugin/systemd-journal-watcher.c new
+377
@@ -0,0 +1,377 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#include "systemd-internals.h"
4 +#include <sys/inotify.h>
5 +
6 +#define EVENT_SIZE (sizeof(struct inotify_event))
7 +#define INITIAL_WATCHES 256
8 +
9 +#define WATCH_FOR (IN_CREATE | IN_MODIFY | IN_DELETE | IN_DELETE_SELF | IN_MOVED_FROM | IN_MOVED_TO | IN_UNMOUNT)
10 +
11 +typedef struct watch_entry {
12 + int slot;
13 +
14 + int wd; // Watch descriptor
15 + char *path; // Dynamically allocated path
16 +
17 + struct watch_entry *next; // for the free list
18 +} WatchEntry;
19 +
20 +typedef struct {
21 + WatchEntry *watchList;
22 + WatchEntry *freeList;
23 + int watchCount;
24 + int watchListSize;
25 +
26 + size_t errors;
27 +
28 + DICTIONARY *pending;
29 +} Watcher;
30 +
31 +static WatchEntry *get_slot(Watcher *watcher) {
32 + WatchEntry *t;
33 +
34 + if (watcher->freeList != NULL) {
35 + t = watcher->freeList;
36 + watcher->freeList = t->next;
37 + t->next = NULL;
38 + return t;
39 + }
40 +
41 + if (watcher->watchCount == watcher->watchListSize) {
42 + watcher->watchListSize *= 2;
43 + watcher->watchList = reallocz(watcher->watchList, watcher->watchListSize * sizeof(WatchEntry));
44 + }
45 +
46 + watcher->watchList[watcher->watchCount] = (WatchEntry){
47 + .slot = watcher->watchCount,
48 + .wd = -1,
49 + .path = NULL,
50 + .next = NULL,
51 + };
52 + t = &watcher->watchList[watcher->watchCount];
53 + watcher->watchCount++;
54 +
55 + return t;
56 +}
57 +
58 +static void free_slot(Watcher *watcher, WatchEntry *t) {
59 + t->wd = -1;
60 + freez(t->path);
61 + t->path = NULL;
62 +
63 + // link it to the free list
64 + t->next = watcher->freeList;
65 + watcher->freeList = t;
66 +}
67 +
68 +static int add_watch(Watcher *watcher, int inotifyFd, const char *path) {
69 + WatchEntry *t = get_slot(watcher);
70 +
71 + t->wd = inotify_add_watch(inotifyFd, path, WATCH_FOR);
72 + if (t->wd == -1) {
73 + nd_log(NDLS_COLLECTORS, NDLP_ERR,
74 + "JOURNAL WATCHER: cannot watch directory: '%s'",
75 + path);
76 +
77 + free_slot(watcher, t);
78 +
79 + struct stat info;
80 + if(stat(path, &info) == 0 && S_ISDIR(info.st_mode)) {
81 + // the directory exists, but we failed to add the watch
82 + // increase errors
83 + watcher->errors++;
84 + }
85 + }
86 + else {
87 + t->path = strdupz(path);
88 +
89 + nd_log(NDLS_COLLECTORS, NDLP_DEBUG,
90 + "JOURNAL WATCHER: watching directory: '%s'",
91 + path);
92 +
93 + }
94 + return t->wd;
95 +}
96 +
97 +static void remove_watch(Watcher *watcher, int inotifyFd, int wd) {
98 + int i;
99 + for (i = 0; i < watcher->watchCount; ++i) {
100 + if (watcher->watchList[i].wd == wd) {
101 +
102 + nd_log(NDLS_COLLECTORS, NDLP_DEBUG,
103 + "JOURNAL WATCHER: removing watch from directory: '%s'",
104 + watcher->watchList[i].path);
105 +
106 + inotify_rm_watch(inotifyFd, watcher->watchList[i].wd);
107 + free_slot(watcher, &watcher->watchList[i]);
108 + return;
109 + }
110 + }
111 +
112 + nd_log(NDLS_COLLECTORS, NDLP_WARNING,
113 + "JOURNAL WATCHER: cannot find directory watch %d to remove.",
114 + wd);
115 +}
116 +
117 +static void free_watches(Watcher *watcher, int inotifyFd) {
118 + for (int i = 0; i < watcher->watchCount; ++i) {
119 + if (watcher->watchList[i].wd != -1) {
120 + inotify_rm_watch(inotifyFd, watcher->watchList[i].wd);
121 + free_slot(watcher, &watcher->watchList[i]);
122 + }
123 + }
124 + freez(watcher->watchList);
125 + watcher->watchList = NULL;
126 +
127 + dictionary_destroy(watcher->pending);
128 + watcher->pending = NULL;
129 +}
130 +
131 +static char* get_path_from_wd(Watcher *watcher, int wd) {
132 + for (int i = 0; i < watcher->watchCount; ++i) {
133 + if (watcher->watchList[i].wd == wd)
134 + return watcher->watchList[i].path;
135 + }
136 + return NULL;
137 +}
138 +
139 +static void watch_directory_recursively(Watcher *watcher, int inotifyFd, const char *basePath) {
140 + // First, add a watch for the top-level directory itself
141 + add_watch(watcher, inotifyFd, basePath);
142 +
143 + char path[PATH_MAX];
144 + struct dirent *dp;
145 + DIR *dir = opendir(basePath);
146 +
147 + if (!dir)
148 + return;
149 +
150 + while ((dp = readdir(dir)) != NULL) {
151 + if (strcmp(dp->d_name, ".") != 0 && strcmp(dp->d_name, "..") != 0) {
152 + snprintfz(path, sizeof(path), "%s/%s", basePath, dp->d_name);
153 +
154 + if (dp->d_type == DT_DIR) {
155 + // Recursively watch this directory
156 + watch_directory_recursively(watcher, inotifyFd, path);
157 + }
158 + }
159 + }
160 +
161 + closedir(dir);
162 +}
163 +
164 +static bool is_subpath(const char *path, const char *subpath) {
165 + // Use strncmp to compare the paths
166 + if (strncmp(path, subpath, strlen(path)) == 0) {
167 + // Ensure that the next character is a '/' or '\0'
168 + char next_char = subpath[strlen(path)];
169 + return next_char == '/' || next_char == '\0';
170 + }
171 +
172 + return false;
173 +}
174 +
175 +void remove_directory_watch(Watcher *watcher, int inotifyFd, const char *dirPath) {
176 + for (int i = 0; i < watcher->watchCount; ++i) {
177 + WatchEntry *t = &watcher->watchList[i];
178 + if (t->wd != -1 && is_subpath(t->path, dirPath)) {
179 + inotify_rm_watch(inotifyFd, t->wd);
180 + free_slot(watcher, t);
181 + }
182 + }
183 +
184 + struct journal_file *jf;
185 + dfe_start_write(journal_files_registry, jf) {
186 + if(is_subpath(jf->filename, dirPath))
187 + dictionary_del(journal_files_registry, jf->filename);
188 + }
189 + dfe_done(jf);
190 +
191 + dictionary_garbage_collect(journal_files_registry);
192 +}
193 +
194 +void process_event(Watcher *watcher, int inotifyFd, struct inotify_event *event) {
195 + if(!event->len) {
196 + nd_log(NDLS_COLLECTORS, NDLP_NOTICE
197 + , "JOURNAL WATCHER: received event with mask %u and len %u (this is zero) for path: '%s' - ignoring it."
198 + , event->mask, event->len, event->name);
199 + return;
200 + }
201 +
202 + char *dirPath = get_path_from_wd(watcher, event->wd);
203 + if(!dirPath) {
204 + nd_log(NDLS_COLLECTORS, NDLP_NOTICE,
205 + "JOURNAL WATCHER: received event with mask %u and len %u for path: '%s' - "
206 + "but we can't find its watch descriptor - ignoring it."
207 + , event->mask, event->len, event->name);
208 + return;
209 + }
210 +
211 + if(event->mask & IN_DELETE_SELF) {
212 + remove_watch(watcher, inotifyFd, event->wd);
213 + return;
214 + }
215 +
216 + static __thread char fullPath[PATH_MAX];
217 + snprintfz(fullPath, sizeof(fullPath), "%s/%s", dirPath, event->name);
218 + // fullPath contains the full path to the file
219 +
220 + size_t len = strlen(event->name);
221 +
222 + if(event->mask & IN_ISDIR) {
223 + if (event->mask & (IN_DELETE | IN_MOVED_FROM)) {
224 + // A directory is deleted or moved out
225 + nd_log(NDLS_COLLECTORS, NDLP_DEBUG,
226 + "JOURNAL WATCHER: Directory deleted or moved out: '%s'",
227 + fullPath);
228 +
229 + // Remove the watch - implement this function based on how you manage your watches
230 + remove_directory_watch(watcher, inotifyFd, fullPath);
231 + }
232 + else if (event->mask & (IN_CREATE | IN_MOVED_TO)) {
233 + // A new directory is created or moved in
234 + nd_log(NDLS_COLLECTORS, NDLP_DEBUG,
235 + "JOURNAL WATCHER: New directory created or moved in: '%s'",
236 + fullPath);
237 +
238 + // Start watching the new directory - recursive watch
239 + watch_directory_recursively(watcher, inotifyFd, fullPath);
240 + }
241 + else
242 + nd_log(NDLS_COLLECTORS, NDLP_WARNING,
243 + "JOURNAL WATCHER: Received unhandled event with mask %u for directory '%s'",
244 + event->mask, fullPath);
245 + }
246 + else if(len > sizeof(".journal") - 1 && strcmp(&event->name[len - (sizeof(".journal") - 1)], ".journal") == 0) {
247 + // It is a file that ends in .journal
248 + // add it to our pending list
249 + dictionary_set(watcher->pending, fullPath, NULL, 0);
250 + }
251 + else
252 + nd_log(NDLS_COLLECTORS, NDLP_DEBUG,
253 + "JOURNAL WATCHER: ignoring event with mask %u for file '%s'",
254 + event->mask, fullPath);
255 +}
256 +
257 +static void process_pending(Watcher *watcher) {
258 + void *x;
259 + dfe_start_write(watcher->pending, x) {
260 + struct stat info;
261 + const char *fullPath = x_dfe.name;
262 +
263 + if(stat(fullPath, &info) != 0) {
264 + nd_log(NDLS_COLLECTORS, NDLP_DEBUG,
265 + "JOURNAL WATCHER: file '%s' no longer exists, removing it from the registry",
266 + fullPath);
267 +
268 + dictionary_del(journal_files_registry, fullPath);
269 + }
270 + else if(S_ISREG(info.st_mode)) {
271 + nd_log(NDLS_COLLECTORS, NDLP_DEBUG,
272 + "JOURNAL WATCHER: file '%s' has been added/updated, updating the registry",
273 + fullPath);
274 +
275 + struct journal_file t = {
276 + .file_last_modified_ut = info.st_mtim.tv_sec * USEC_PER_SEC +
277 + info.st_mtim.tv_nsec / NSEC_PER_USEC,
278 + .last_scan_monotonic_ut = now_monotonic_usec(),
279 + .size = info.st_size,
280 + .max_journal_vs_realtime_delta_ut = JOURNAL_VS_REALTIME_DELTA_DEFAULT_UT,
281 + };
282 + struct journal_file *jf = dictionary_set(journal_files_registry, fullPath, &t, sizeof(t));
283 + journal_file_update_header(jf->filename, jf);
284 + }
285 +
286 + dictionary_del(watcher->pending, fullPath);
287 + }
288 + dfe_done(x);
289 +
290 + dictionary_garbage_collect(watcher->pending);
291 +}
292 +
293 +void *journal_watcher_main(void *arg __maybe_unused) {
294 + while(1) {
295 + Watcher watcher = {
296 + .watchList = mallocz(INITIAL_WATCHES * sizeof(WatchEntry)),
297 + .freeList = NULL,
298 + .watchCount = 0,
299 + .watchListSize = INITIAL_WATCHES,
300 + .pending = dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE|DICT_OPTION_SINGLE_THREADED),
301 + .errors = 0,
302 + };
303 +
304 + int inotifyFd = inotify_init();
305 + if (inotifyFd < 0) {
306 + nd_log(NDLS_COLLECTORS, NDLP_ERR, "inotify_init() failed.");
307 + free_watches(&watcher, inotifyFd);
308 + return NULL;
309 + }
310 +
311 + for (unsigned i = 0; i < MAX_JOURNAL_DIRECTORIES; i++) {
312 + if (!journal_directories[i].path) break;
313 + watch_directory_recursively(&watcher, inotifyFd, journal_directories[i].path);
314 + }
315 +
316 + usec_t last_headers_update_ut = now_monotonic_usec();
317 + struct buffered_reader reader;
318 + while (1) {
319 + buffered_reader_ret_t rc = buffered_reader_read_timeout(
320 + &reader, inotifyFd, EXECUTE_WATCHER_PENDING_EVERY_MS, false);
321 +
322 + if (rc != BUFFERED_READER_READ_OK && rc != BUFFERED_READER_READ_POLL_TIMEOUT) {
323 + nd_log(NDLS_COLLECTORS, NDLP_CRIT,
324 + "JOURNAL WATCHER: cannot read inotify events, buffered_reader_read_timeout() returned %d - "
325 + "restarting the watcher.",
326 + rc);
327 + break;
328 + }
329 +
330 + if(rc == BUFFERED_READER_READ_OK) {
331 + bool unmount_event = false;
332 +
333 + ssize_t i = 0;
334 + while (i < reader.read_len) {
335 + struct inotify_event *event = (struct inotify_event *) &reader.read_buffer[i];
336 +
337 + if(event->mask & IN_UNMOUNT) {
338 + unmount_event = true;
339 + break;
340 + }
341 +
342 + process_event(&watcher, inotifyFd, event);
343 + i += (ssize_t)EVENT_SIZE + event->len;
344 + }
345 +
346 + reader.read_buffer[0] = '\0';
347 + reader.read_len = 0;
348 + reader.pos = 0;
349 +
350 + if(unmount_event)
351 + break;
352 + }
353 +
354 + usec_t ut = now_monotonic_usec();
355 + if (dictionary_entries(watcher.pending) && (rc == BUFFERED_READER_READ_POLL_TIMEOUT ||
356 + last_headers_update_ut + (EXECUTE_WATCHER_PENDING_EVERY_MS * USEC_PER_MS) <= ut)) {
357 + process_pending(&watcher);
358 + last_headers_update_ut = ut;
359 + }
360 +
361 + if(watcher.errors) {
362 + nd_log(NDLS_COLLECTORS, NDLP_NOTICE,
363 + "JOURNAL WATCHER: there were errors in setting up inotify watches - restarting the watcher.");
364 + }
365 + }
366 +
367 + close(inotifyFd);
368 + free_watches(&watcher, inotifyFd);
369 +
370 + // this will scan the directories and cleanup the registry
371 + journal_files_registry_update();
372 +
373 + sleep_usec(5 * USEC_PER_SEC);
374 + }
375 +
376 + return NULL;
377 +}
collectors/systemd-journal.plugin/systemd-journal.c
+117 -68
@@ -215,6 +215,9 @@ typedef struct function_query_status {
215 usec_t start_ut; // the starting time of the query - we start from this
216 usec_t stop_ut; // the ending time of the query - we stop at this
217 usec_t first_msg_ut;
218 +
219 + sd_id128_t first_msg_writer;
220 + uint64_t first_msg_seqnum;
221 } query_file;
222
223 struct {
@@ -360,19 +363,20 @@ static void sampling_file_init(FUNCTION_QUERY_STATUS *fqs, struct journal_file *
363 fqs->samples_per_file.recalibrate = 0;
364 }
365
363 -static size_t sampling_file_lines_scanned(FUNCTION_QUERY_STATUS *fqs) {
366 +static size_t sampling_file_lines_scanned_so_far(FUNCTION_QUERY_STATUS *fqs) {
367 size_t sampled = fqs->samples_per_file.sampled + fqs->samples_per_file.unsampled;
368 if(!sampled) sampled = 1;
369 return sampled;
370 }
371
369 -static double sampling_file_query_overlapping_timeframe_ut(FUNCTION_QUERY_STATUS *fqs, struct journal_file *jf, FACETS_ANCHOR_DIRECTION direction,
370 - usec_t msg_ut, usec_t *after_ut, usec_t *before_ut) {
372 +static void sampling_running_file_query_overlapping_timeframe_ut(
373 + FUNCTION_QUERY_STATUS *fqs, struct journal_file *jf, FACETS_ANCHOR_DIRECTION direction,
374 + usec_t msg_ut, usec_t *after_ut, usec_t *before_ut) {
375 +
376 // find the overlap of the query and file timeframes
377 // taking into account the first message we encountered
378
374 - double progress;
375 - usec_t oldest_ut, newest_ut, elapsed_ut;
379 + usec_t oldest_ut, newest_ut;
380 if(direction == FACETS_ANCHOR_DIRECTION_FORWARD) {
381 // the first message we know (oldest)
382 oldest_ut = fqs->query_file.first_msg_ut ? fqs->query_file.first_msg_ut : jf->msg_first_ut;
@@ -387,8 +391,6 @@ static double sampling_file_query_overlapping_timeframe_ut(FUNCTION_QUERY_STATUS
391
392 if(msg_ut < oldest_ut)
393 oldest_ut = msg_ut - 1;
390 -
391 - elapsed_ut = msg_ut - oldest_ut;
394 }
395 else /* BACKWARD */ {
396 // the latest message we know (newest)
@@ -402,24 +404,35 @@ static double sampling_file_query_overlapping_timeframe_ut(FUNCTION_QUERY_STATUS
404
405 if(newest_ut < msg_ut)
406 newest_ut = msg_ut + 1;
405 -
406 - elapsed_ut = newest_ut - msg_ut;
407 }
408
409 - usec_t total_ut = newest_ut - oldest_ut;
410 - progress = (double)elapsed_ut / (double)total_ut;
411 -
409 *after_ut = oldest_ut;
410 *before_ut = newest_ut;
411 +}
412 +
413 +static double sampling_running_file_query_progress_by_time(FUNCTION_QUERY_STATUS *fqs, struct journal_file *jf,
414 + FACETS_ANCHOR_DIRECTION direction, usec_t msg_ut) {
415 +
416 + usec_t after_ut, before_ut, elapsed_ut;
417 + sampling_running_file_query_overlapping_timeframe_ut(fqs, jf, direction, msg_ut, &after_ut, &before_ut);
418 +
419 + if(direction == FACETS_ANCHOR_DIRECTION_FORWARD)
420 + elapsed_ut = msg_ut - after_ut;
421 + else
422 + elapsed_ut = before_ut - msg_ut;
423 +
424 + usec_t total_ut = before_ut - after_ut;
425 + double progress = (double)elapsed_ut / (double)total_ut;
426
427 return progress;
428 }
429
418 -static usec_t sampling_file_remaining_time_ut(FUNCTION_QUERY_STATUS *fqs, struct journal_file *jf, FACETS_ANCHOR_DIRECTION direction,
419 - usec_t msg_ut, usec_t *total_time_ut, usec_t *remaining_start_ut, usec_t *remaining_end_ut) {
420 -
430 +static usec_t sampling_running_file_query_remaining_time(FUNCTION_QUERY_STATUS *fqs, struct journal_file *jf,
431 + FACETS_ANCHOR_DIRECTION direction, usec_t msg_ut,
432 + usec_t *total_time_ut, usec_t *remaining_start_ut,
433 + usec_t *remaining_end_ut) {
434 usec_t after_ut, before_ut;
422 - sampling_file_query_overlapping_timeframe_ut(fqs, jf, direction, msg_ut, &after_ut, &before_ut);
435 + sampling_running_file_query_overlapping_timeframe_ut(fqs, jf, direction, msg_ut, &after_ut, &before_ut);
436
437 // since we have a timestamp in msg_ut
438 // this timestamp can extend the overlap
@@ -454,38 +467,84 @@ static usec_t sampling_file_remaining_time_ut(FUNCTION_QUERY_STATUS *fqs, struct
467 return remaining_ut;
468 }
469
457 -static size_t sampling_file_estimate_remaining_lines(FUNCTION_QUERY_STATUS *fqs, struct journal_file *jf, FACETS_ANCHOR_DIRECTION direction, usec_t msg_ut) {
458 - size_t scanned_lines = sampling_file_lines_scanned(fqs);
470 +static size_t sampling_running_file_query_estimate_remaining_lines_by_time(FUNCTION_QUERY_STATUS *fqs,
471 + struct journal_file *jf,
472 + FACETS_ANCHOR_DIRECTION direction,
473 + usec_t msg_ut) {
474 + size_t scanned_lines = sampling_file_lines_scanned_so_far(fqs);
475
476 + // Calculate the proportion of time covered
477 usec_t total_time_ut;
461 - usec_t remaining_time_ut = sampling_file_remaining_time_ut(fqs, jf, direction, msg_ut, &total_time_ut, NULL, NULL);
462 -
463 - if (total_time_ut == 0)
464 - total_time_ut = 1;
478 + usec_t remaining_time_ut = sampling_running_file_query_remaining_time(fqs, jf, direction, msg_ut, &total_time_ut,
479 + NULL, NULL);
480 + if (total_time_ut == 0) total_time_ut = 1;
481
466 - // Calculate the proportion of time covered
467 - double time_proportion = (double)(total_time_ut - remaining_time_ut) / (double)total_time_ut;
482 + double proportion_by_time = (double) (total_time_ut - remaining_time_ut) / (double) total_time_ut;
483
469 - if (time_proportion == 0 || !isfinite(time_proportion))
470 - time_proportion = 1.0;
484 + if (proportion_by_time == 0 || proportion_by_time > 1.0 || !isfinite(proportion_by_time))
485 + proportion_by_time = 1.0;
486
487 // Estimate the total number of lines in the file
473 - size_t total_lines_estimated = (size_t)((double)scanned_lines / time_proportion);
488 + size_t expected_matching_logs_by_time = (size_t)((double)scanned_lines / proportion_by_time);
489 +
490 + if(jf->messages_in_file && expected_matching_logs_by_time > jf->messages_in_file)
491 + expected_matching_logs_by_time = jf->messages_in_file;
492
493 // Calculate the estimated number of remaining lines
476 - size_t expected_lines = total_lines_estimated - scanned_lines;
494 + size_t remaining_logs_by_time = expected_matching_logs_by_time - scanned_lines;
495 + if (remaining_logs_by_time < 1) remaining_logs_by_time = 1;
496 +
497 + return remaining_logs_by_time;
498 +}
499
478 - if (expected_lines < 1)
479 - expected_lines = 1;
500
481 - return expected_lines;
501 +static size_t sampling_running_file_query_estimate_remaining_lines(sd_journal *j, FUNCTION_QUERY_STATUS *fqs, struct journal_file *jf, FACETS_ANCHOR_DIRECTION direction, usec_t msg_ut) {
502 + size_t expected_matching_logs_by_seqnum = 0;
503 + double proportion_by_seqnum = 0.0;
504 + size_t remaining_logs_by_seqnum = 0;
505 +
506 +#ifdef HAVE_SD_JOURNAL_GET_SEQNUM
507 + uint64_t current_msg_seqnum;
508 + sd_id128_t current_msg_writer;
509 + if(!fqs->query_file.first_msg_seqnum || sd_journal_get_seqnum(j, &current_msg_seqnum, &current_msg_writer) < 0) {
510 + fqs->query_file.first_msg_seqnum = 0;
511 + fqs->query_file.first_msg_writer = SD_ID128_NULL;
512 + }
513 + else if(jf->messages_in_file) {
514 + size_t scanned_lines = sampling_file_lines_scanned_so_far(fqs);
515 +
516 + double proportion_of_all_lines_so_far;
517 + if(direction == FACETS_ANCHOR_DIRECTION_FORWARD)
518 + proportion_of_all_lines_so_far = (double)scanned_lines / (double)(current_msg_seqnum - jf->first_seqnum);
519 + else
520 + proportion_of_all_lines_so_far = (double)scanned_lines / (double)(jf->last_seqnum - current_msg_seqnum);
521 +
522 + if(proportion_of_all_lines_so_far > 1.0)
523 + proportion_of_all_lines_so_far = 1.0;
524 +
525 + expected_matching_logs_by_seqnum = (size_t)(proportion_of_all_lines_so_far * (double)jf->messages_in_file);
526 +
527 + proportion_by_seqnum = (double)scanned_lines / (double)expected_matching_logs_by_seqnum;
528 +
529 + if (proportion_by_seqnum == 0 || proportion_by_seqnum > 1.0 || !isfinite(proportion_by_seqnum))
530 + proportion_by_seqnum = 1.0;
531 +
532 + remaining_logs_by_seqnum = expected_matching_logs_by_seqnum - scanned_lines;
533 + if(!remaining_logs_by_seqnum) remaining_logs_by_seqnum = 1;
534 + }
535 +#endif
536 +
537 + if(remaining_logs_by_seqnum)
538 + return remaining_logs_by_seqnum;
539 +
540 + return sampling_running_file_query_estimate_remaining_lines_by_time(fqs, jf, direction, msg_ut);
541 }
542
484 -static void sampling_decide_file_sampling_every(FUNCTION_QUERY_STATUS *fqs, struct journal_file *jf, FACETS_ANCHOR_DIRECTION direction, usec_t msg_ut) {
543 +static void sampling_decide_file_sampling_every(sd_journal *j, FUNCTION_QUERY_STATUS *fqs, struct journal_file *jf, FACETS_ANCHOR_DIRECTION direction, usec_t msg_ut) {
544 size_t files_matched = fqs->files_matched;
545 if(!files_matched) files_matched = 1;
546
488 - size_t remaining_lines = sampling_file_estimate_remaining_lines(fqs, jf, direction, msg_ut);
547 + size_t remaining_lines = sampling_running_file_query_estimate_remaining_lines(j, fqs, jf, direction, msg_ut);
548 size_t wanted_samples = (fqs->sampling / 2) / files_matched;
549 if(!wanted_samples) wanted_samples = 1;
550
@@ -501,7 +560,7 @@ typedef enum {
560 SAMPLING_SKIP_FIELDS = 1,
561 } sampling_t;
562
504 -static inline sampling_t is_row_in_sample(FUNCTION_QUERY_STATUS *fqs, struct journal_file *jf, usec_t msg_ut, FACETS_ANCHOR_DIRECTION direction, bool candidate_to_keep) {
563 +static inline sampling_t is_row_in_sample(sd_journal *j, FUNCTION_QUERY_STATUS *fqs, struct journal_file *jf, usec_t msg_ut, FACETS_ANCHOR_DIRECTION direction, bool candidate_to_keep) {
564 if(!fqs->sampling || candidate_to_keep)
565 return SAMPLING_FULL;
566
@@ -523,7 +582,7 @@ static inline sampling_t is_row_in_sample(FUNCTION_QUERY_STATUS *fqs, struct jou
582
583 else if(fqs->samples_per_file.recalibrate >= SYSTEMD_JOURNAL_SAMPLING_RECALIBRATE || !fqs->samples_per_file.every) {
584 // this is the first to be unsampled for this file
526 - sampling_decide_file_sampling_every(fqs, jf, direction, msg_ut);
585 + sampling_decide_file_sampling_every(j, fqs, jf, direction, msg_ut);
586 fqs->samples_per_file.recalibrate = 0;
587 should_sample = true;
588 }
@@ -552,22 +611,20 @@ static inline sampling_t is_row_in_sample(FUNCTION_QUERY_STATUS *fqs, struct jou
611 fqs->samples_per_time_slot.unsampled[slot]++;
612
613 if(fqs->samples_per_file.unsampled > fqs->samples_per_file.sampled) {
555 - usec_t after_ut, before_ut;
556 -
557 - double progress = sampling_file_query_overlapping_timeframe_ut(
558 - fqs, jf, direction, msg_ut, &after_ut, &before_ut);
614 + double progress_by_time = sampling_running_file_query_progress_by_time(fqs, jf, direction, msg_ut);
615
560 - if(progress > 0.01)
616 + if(progress_by_time > 0.05)
617 return SAMPLING_STOP_AND_ESTIMATE;
618 }
619
620 return SAMPLING_SKIP_FIELDS;
621 }
622
567 -static void sampling_update_file_estimates(FACETS *facets, FUNCTION_QUERY_STATUS *fqs, struct journal_file *jf, usec_t msg_ut, FACETS_ANCHOR_DIRECTION direction) {
623 +static void sampling_update_running_query_file_estimates(FACETS *facets, sd_journal *j, FUNCTION_QUERY_STATUS *fqs, struct journal_file *jf, usec_t msg_ut, FACETS_ANCHOR_DIRECTION direction) {
624 usec_t total_time_ut, remaining_start_ut, remaining_end_ut;
569 - sampling_file_remaining_time_ut(fqs, jf, direction, msg_ut, &total_time_ut, &remaining_start_ut, &remaining_end_ut);
570 - size_t remaining_lines = sampling_file_estimate_remaining_lines(fqs, jf, direction, msg_ut);
625 + sampling_running_file_query_remaining_time(fqs, jf, direction, msg_ut, &total_time_ut, &remaining_start_ut,
626 + &remaining_end_ut);
627 + size_t remaining_lines = sampling_running_file_query_estimate_remaining_lines(j, fqs, jf, direction, msg_ut);
628 facets_update_estimations(facets, remaining_start_ut, remaining_end_ut, remaining_lines);
629 fqs->samples.estimated += remaining_lines;
630 fqs->samples_per_file.estimated += remaining_lines;
@@ -575,25 +632,6 @@ static void sampling_update_file_estimates(FACETS *facets, FUNCTION_QUERY_STATUS
632
633 // ----------------------------------------------------------------------------
634
578 -static inline bool parse_journal_field(const char *data, size_t data_length, const char **key, size_t *key_length, const char **value, size_t *value_length) {
579 - const char *k = data;
580 - const char *equal = strchr(k, '=');
581 - if(unlikely(!equal))
582 - return false;
583 -
584 - size_t kl = equal - k;
585 -
586 - const char *v = ++equal;
587 - size_t vl = data_length - kl - 1;
588 -
589 - *key = k;
590 - *key_length = kl;
591 - *value = v;
592 - *value_length = vl;
593 -
594 - return true;
595 -}
596 -
635 static inline size_t netdata_systemd_journal_process_row(sd_journal *j, FACETS *facets, struct journal_file *jf, usec_t *msg_ut) {
636 const void *data;
637 size_t length, bytes = 0;
@@ -707,9 +745,16 @@ ND_SD_JOURNAL_STATUS netdata_systemd_journal_query_backward(
745 if(unlikely(!first_msg_ut)) {
746 first_msg_ut = msg_ut;
747 fqs->query_file.first_msg_ut = msg_ut;
748 +
749 +#ifdef HAVE_SD_JOURNAL_GET_SEQNUM
750 + if(sd_journal_get_seqnum(j, &fqs->query_file.first_msg_seqnum, &fqs->query_file.first_msg_writer) < 0) {
751 + fqs->query_file.first_msg_seqnum = 0;
752 + fqs->query_file.first_msg_writer = SD_ID128_NULL;
753 + }
754 +#endif
755 }
756
712 - sampling_t sample = is_row_in_sample(fqs, jf, msg_ut,
757 + sampling_t sample = is_row_in_sample(j, fqs, jf, msg_ut,
758 FACETS_ANCHOR_DIRECTION_BACKWARD,
759 facets_row_candidate_to_keep(facets, msg_ut));
760
@@ -748,7 +793,7 @@ ND_SD_JOURNAL_STATUS netdata_systemd_journal_query_backward(
793 else if(sample == SAMPLING_SKIP_FIELDS)
794 facets_row_finished_unsampled(facets, msg_ut);
795 else {
751 - sampling_update_file_estimates(facets, fqs, jf, msg_ut, FACETS_ANCHOR_DIRECTION_BACKWARD);
796 + sampling_update_running_query_file_estimates(facets, j, fqs, jf, msg_ut, FACETS_ANCHOR_DIRECTION_BACKWARD);
797 break;
798 }
799 }
@@ -816,7 +861,7 @@ ND_SD_JOURNAL_STATUS netdata_systemd_journal_query_forward(
861 fqs->query_file.first_msg_ut = msg_ut;
862 }
863
819 - sampling_t sample = is_row_in_sample(fqs, jf, msg_ut,
864 + sampling_t sample = is_row_in_sample(j, fqs, jf, msg_ut,
865 FACETS_ANCHOR_DIRECTION_FORWARD,
866 facets_row_candidate_to_keep(facets, msg_ut));
867
@@ -855,7 +900,7 @@ ND_SD_JOURNAL_STATUS netdata_systemd_journal_query_forward(
900 else if(sample == SAMPLING_SKIP_FIELDS)
901 facets_row_finished_unsampled(facets, msg_ut);
902 else {
858 - sampling_update_file_estimates(facets, fqs, jf, msg_ut, FACETS_ANCHOR_DIRECTION_FORWARD);
903 + sampling_update_running_query_file_estimates(facets, j, fqs, jf, msg_ut, FACETS_ANCHOR_DIRECTION_FORWARD);
904 break;
905 }
906 }
@@ -1020,8 +1065,13 @@ static bool jf_is_mine(struct journal_file *jf, FUNCTION_QUERY_STATUS *fqs) {
1065 if((fqs->source_type == SDJF_NONE && !fqs->sources) || (jf->source_type & fqs->source_type) ||
1066 (fqs->sources && simple_pattern_matches(fqs->sources, string2str(jf->source)))) {
1067
1068 + if(!jf->msg_last_ut || !jf->msg_last_ut)
1069 + // the file is not scanned yet, or the timestamps have not been updated,
1070 + // so we don't know if it can contribute or not - let's add it.
1071 + return true;
1072 +
1073 usec_t anchor_delta = JOURNAL_VS_REALTIME_DELTA_MAX_UT;
1024 - usec_t first_ut = jf->msg_first_ut;
1074 + usec_t first_ut = jf->msg_first_ut - anchor_delta;
1075 usec_t last_ut = jf->msg_last_ut + anchor_delta;
1076
1077 if(last_ut >= fqs->after_ut && first_ut <= fqs->before_ut)
@@ -1503,7 +1553,6 @@ static void function_systemd_journal_progress(BUFFER *wb, const char *transactio
1553 void function_systemd_journal(const char *transaction, char *function, int timeout, bool *cancelled) {
1554 fstat_thread_calls = 0;
1555 fstat_thread_cached_responses = 0;
1506 - journal_files_registry_update();
1556
1557 BUFFER *wb = buffer_create(0, NULL);
1558 buffer_flush(wb);
collectors/systemd-journal.plugin/systemd-main.c
+12 -2
@@ -10,6 +10,7 @@ static bool plugin_should_exit = false;
10
11 int main(int argc __maybe_unused, char **argv __maybe_unused) {
12 clocks_init();
13 + netdata_thread_set_tag("SDMAIN");
14 nd_log_initialize_for_external_plugins("systemd-journal.plugin");
15
16 netdata_configured_host_prefix = getenv("NETDATA_HOST_PREFIX");
@@ -26,6 +27,8 @@ int main(int argc __maybe_unused, char **argv __maybe_unused) {
27 // debug
28
29 if(argc == 2 && strcmp(argv[1], "debug") == 0) {
30 + journal_files_registry_update();
31 +
32 bool cancelled = false;
33 char buf[] = "systemd-journal after:-8640000 before:0 direction:backward last:200 data_only:false slice:true source:all";
34 // char buf[] = "systemd-journal after:1695332964 before:1695937764 direction:backward last:100 slice:true source:all DHKucpqUoe1:PtVoyIuX.MU";
@@ -42,6 +45,13 @@ int main(int argc __maybe_unused, char **argv __maybe_unused) {
45 }
46 #endif
47
48 + // ------------------------------------------------------------------------
49 + // watcher thread
50 +
51 + netdata_thread_t watcher_thread;
52 + netdata_thread_create(&watcher_thread, "SDWATCH",
53 + NETDATA_THREAD_OPTION_DONT_LOG, journal_watcher_main, NULL);
54 +
55 // ------------------------------------------------------------------------
56 // the event loop for functions
57
@@ -76,14 +86,14 @@ int main(int argc __maybe_unused, char **argv __maybe_unused) {
86
87 usec_t step_ut = 100 * USEC_PER_MS;
88 usec_t send_newline_ut = 0;
79 - usec_t since_last_scan_ut = 1000 * USEC_PER_SEC; // something big to trigger scanning at start
89 + usec_t since_last_scan_ut = FULL_JOURNAL_SCAN_EVERY_USEC * 2; // something big to trigger scanning at start
90 bool tty = isatty(fileno(stderr)) == 1;
91
92 heartbeat_t hb;
93 heartbeat_init(&hb);
94 while(!plugin_should_exit) {
95
86 - if(since_last_scan_ut > 60 * USEC_PER_SEC) {
96 + if(since_last_scan_ut > FULL_JOURNAL_SCAN_EVERY_USEC) {
97 journal_files_registry_update();
98 since_last_scan_ut = 0;
99 }
configure.ac
+5
@@ -1231,6 +1231,11 @@ if test "${have_sd_journal_restart_fields}" = "yes"; then
1231 AC_DEFINE([HAVE_SD_JOURNAL_RESTART_FIELDS], [1], [sd_journal_restart_fields usability])
1232 fi
1233
1234 +AC_CHECK_LIB([systemd], [sd_journal_get_seqnum], [have_sd_journal_get_seqnum=yes], [have_sd_journal_get_seqnum=no])
1235 +if test "${have_sd_journal_get_seqnum}" = "yes"; then
1236 + AC_DEFINE([HAVE_SD_JOURNAL_GET_SEQNUM], [1], [sd_journal_get_seqnum usability])
1237 +fi
1238 +
1239 AC_CHECK_LIB([systemd], [sd_bus_default_system, sd_bus_call_method, sd_bus_message_enter_container, sd_bus_message_read, sd_bus_message_exit_container],
1240 [SYSTEMD_DBUS_FOUND=yes],
1241 [SYSTEMD_DBUS_FOUND=no])
libnetdata/dictionary/dictionary.c
+1 -1
@@ -422,7 +422,7 @@ static inline void DICTIONARY_ENTRIES_MINUS1(DICTIONARY *dict) {
422 size_t entries; (void)entries;
423 if(unlikely(is_dictionary_single_threaded(dict))) {
424 dict->version++;
425 - entries = dict->entries++;
425 + entries = dict->entries--;
426 }
427 else {
428 __atomic_fetch_add(&dict->version, 1, __ATOMIC_RELAXED);
libnetdata/libnetdata.c
+24
@@ -1696,6 +1696,30 @@ char *find_and_replace(const char *src, const char *find, const char *replace, c
1696 return value;
1697 }
1698
1699 +
1700 +BUFFER *run_command_and_get_output_to_buffer(const char *command, int max_line_length) {
1701 + BUFFER *wb = buffer_create(0, NULL);
1702 +
1703 + pid_t pid;
1704 + FILE *fp = netdata_popen(command, &pid, NULL);
1705 +
1706 + if(fp) {
1707 + char buffer[max_line_length + 1];
1708 + while (fgets(buffer, max_line_length, fp)) {
1709 + buffer[max_line_length] = '\0';
1710 + buffer_strcat(wb, buffer);
1711 + }
1712 + }
1713 + else {
1714 + buffer_free(wb);
1715 + netdata_log_error("Failed to execute command '%s'.", command);
1716 + return NULL;
1717 + }
1718 +
1719 + netdata_pclose(NULL, fp, pid);
1720 + return wb;
1721 +}
1722 +
1723 bool run_command_and_copy_output_to_stdout(const char *command, int max_line_length) {
1724 pid_t pid;
1725 FILE *fp = netdata_popen(command, &pid, NULL);
libnetdata/libnetdata.h
+1
@@ -689,6 +689,7 @@ static inline BITMAPX *bitmapX_create(uint32_t bits) {
689 #define PLUGINSD_LINE_MAX (COMPRESSION_MAX_MSG_SIZE - 768)
690
691 bool run_command_and_copy_output_to_stdout(const char *command, int max_line_length);
692 +struct web_buffer *run_command_and_get_output_to_buffer(const char *command, int max_line_length);
693
694 typedef enum {
695 OPEN_FD_ACTION_CLOSE,
libnetdata/threads/threads.c
+16 -13
@@ -203,27 +203,30 @@ static void thread_cleanup(void *ptr) {
203 netdata_thread = NULL;
204 }
205
206 -static void thread_set_name_np(NETDATA_THREAD *nt) {
206 +void netdata_thread_set_tag(const char *tag) {
207 + if(!tag || !*tag)
208 + return;
209
208 - if (nt && nt->tag[0]) {
209 - int ret = 0;
210 + int ret = 0;
211
211 - char threadname[NETDATA_THREAD_NAME_MAX+1];
212 - strncpyz(threadname, nt->tag, NETDATA_THREAD_NAME_MAX);
212 + char threadname[NETDATA_THREAD_NAME_MAX+1];
213 + strncpyz(threadname, tag, NETDATA_THREAD_NAME_MAX);
214
215 #if defined(__FreeBSD__)
215 - pthread_set_name_np(pthread_self(), threadname);
216 + pthread_set_name_np(pthread_self(), threadname);
217 #elif defined(__APPLE__)
217 - ret = pthread_setname_np(threadname);
218 + ret = pthread_setname_np(threadname);
219 #else
219 - ret = pthread_setname_np(pthread_self(), threadname);
220 + ret = pthread_setname_np(pthread_self(), threadname);
221 #endif
222
222 - if (ret != 0)
223 - nd_log(NDLS_DAEMON, NDLP_WARNING, "cannot set pthread name of %d to %s. ErrCode: %d", gettid(), threadname, ret);
224 - else
225 - nd_log(NDLS_DAEMON, NDLP_DEBUG, "set name of thread %d to %s", gettid(), threadname);
223 + if (ret != 0)
224 + nd_log(NDLS_DAEMON, NDLP_WARNING, "cannot set pthread name of %d to %s. ErrCode: %d", gettid(), threadname, ret);
225 + else
226 + nd_log(NDLS_DAEMON, NDLP_DEBUG, "set name of thread %d to %s", gettid(), threadname);
227
228 + if(netdata_thread) {
229 + strncpyz(netdata_thread->tag, threadname, sizeof(netdata_thread->tag) - 1);
230 }
231 }
232
@@ -270,7 +273,7 @@ static void *netdata_thread_init(void *ptr) {
273 if(pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL) != 0)
274 nd_log(NDLS_DAEMON, NDLP_WARNING, "cannot set pthread cancel state to ENABLE.");
275
273 - thread_set_name_np(ptr);
276 + netdata_thread_set_tag(netdata_thread->tag);
277
278 void *ret = NULL;
279 pthread_cleanup_push(thread_cleanup, ptr);
libnetdata/threads/threads.h
+2
@@ -18,6 +18,8 @@ typedef enum {
18 #define netdata_thread_cleanup_push(func, arg) pthread_cleanup_push(func, arg)
19 #define netdata_thread_cleanup_pop(execute) pthread_cleanup_pop(execute)
20
21 +void netdata_thread_set_tag(const char *tag);
22 +
23 typedef pthread_t netdata_thread_t;
24
25 struct netdata_static_thread {