preparation for virtual logs functions (#22584)
Costa Tsaousis committed
May 31, 2026 at 04:01 UTC
7e9cbb5dab6fb32d2a95dd8a3e575dc094dd0c41
6 files changed
+938
-903
CMakeLists.txt
+2
@@ -1840,6 +1840,8 @@ set(SYSTEMD_JOURNAL_PLUGIN_FILES
1840
src/collectors/systemd-journal.plugin/systemd-internals.h
1841
src/collectors/systemd-journal.plugin/systemd-main.c
1842
src/collectors/systemd-journal.plugin/systemd-journal.c
1843
+ src/collectors/systemd-journal.plugin/systemd-journal-function.h
1844
+ src/collectors/systemd-journal.plugin/systemd-journal-execute.h
1845
src/collectors/systemd-journal.plugin/systemd-journal-annotations.c
1846
src/collectors/systemd-journal.plugin/systemd-journal-files.c
1847
src/collectors/systemd-journal.plugin/systemd-journal-watcher.c
src/collectors/systemd-journal.plugin/systemd-journal-execute.h
new
+816
@@ -0,0 +1,816 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#ifndef NETDATA_SYSTEMD_JOURNAL_EXECUTE_H
4
+#define NETDATA_SYSTEMD_JOURNAL_EXECUTE_H
5
+
6
+#define JOURNAL_KEY_ND_JOURNAL_FILE "ND_JOURNAL_FILE"
7
+
8
+#define FUNCTION_PROGRESS_UPDATE_ROWS(rows_read, rows) __atomic_fetch_add(&(rows_read), rows, __ATOMIC_RELAXED)
9
+#define FUNCTION_PROGRESS_UPDATE_BYTES(bytes_read, bytes) __atomic_fetch_add(&(bytes_read), bytes, __ATOMIC_RELAXED)
10
+#define FUNCTION_PROGRESS_EVERY_ROWS (1ULL << 13)
11
+#define FUNCTION_DATA_ONLY_CHECK_EVERY_ROWS (1ULL << 7)
12
+
13
+#include "systemd-journal-sampling.h"
14
+
15
+static __always_inline
16
+bool nd_sd_journal_seek_to(NsdJournal *j, usec_t timestamp)
17
+{
18
+ if (nsd_journal_seek_realtime_usec(j, timestamp) < 0) {
19
+ netdata_log_error("SYSTEMD-JOURNAL: Failed to seek to %" PRIu64, timestamp);
20
+ if (nsd_journal_seek_tail(j) < 0) {
21
+ netdata_log_error("SYSTEMD-JOURNAL: Failed to seek to journal's tail");
22
+ return false;
23
+ }
24
+ }
25
+
26
+ return true;
27
+}
28
+
29
+static __always_inline
30
+size_t nd_sd_journal_process_row(NsdJournal *j, FACETS *facets, struct nd_journal_file *njf, usec_t *msg_ut) {
31
+ const void *data;
32
+ size_t length, bytes = 0;
33
+
34
+ facets_add_key_value_length(
35
+ facets, JOURNAL_KEY_ND_JOURNAL_FILE, sizeof(JOURNAL_KEY_ND_JOURNAL_FILE) - 1, njf->filename, njf->filename_len);
36
+
37
+ NSD_JOURNAL_FOREACH_DATA(j, data, length)
38
+ {
39
+ const char *key, *value;
40
+ size_t key_length, value_length;
41
+
42
+ if (!parse_journal_field(data, length, &key, &key_length, &value, &value_length))
43
+ continue;
44
+
45
+#ifdef NETDATA_INTERNAL_CHECKS
46
+ usec_t origin_journal_ut = *msg_ut;
47
+#endif
48
+ if (unlikely(
49
+ key_length == sizeof(JD_SOURCE_REALTIME_TIMESTAMP) - 1 &&
50
+ memcmp(key, JD_SOURCE_REALTIME_TIMESTAMP, sizeof(JD_SOURCE_REALTIME_TIMESTAMP) - 1) == 0)) {
51
+ usec_t ut = str2ull(value, NULL);
52
+ if (ut && ut < *msg_ut) {
53
+ usec_t delta = *msg_ut - ut;
54
+ *msg_ut = ut;
55
+
56
+ if (delta > JOURNAL_VS_REALTIME_DELTA_MAX_UT)
57
+ delta = JOURNAL_VS_REALTIME_DELTA_MAX_UT;
58
+
59
+ // update max_journal_vs_realtime_delta_ut if the delta increased
60
+ usec_t expected = njf->max_journal_vs_realtime_delta_ut;
61
+ do {
62
+ if (delta <= expected)
63
+ break;
64
+ } while (!__atomic_compare_exchange_n(
65
+ &njf->max_journal_vs_realtime_delta_ut, &expected, delta, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
66
+
67
+ internal_error(
68
+ delta > expected,
69
+ "increased max_journal_vs_realtime_delta_ut from %" PRIu64 " to %" PRIu64 ", "
70
+ "journal %" PRIu64 ", actual %" PRIu64 " (delta %" PRIu64 ")",
71
+ expected,
72
+ delta,
73
+ origin_journal_ut,
74
+ *msg_ut,
75
+ origin_journal_ut - (*msg_ut));
76
+ }
77
+ }
78
+
79
+ bytes += length;
80
+ facets_add_key_value_length(
81
+ facets,
82
+ key,
83
+ key_length,
84
+ value,
85
+ value_length <= FACET_MAX_VALUE_LENGTH ? value_length : FACET_MAX_VALUE_LENGTH);
86
+ }
87
+
88
+ return bytes;
89
+}
90
+
91
+static __always_inline
92
+ND_SD_JOURNAL_STATUS check_stop(const bool *cancelled, const usec_t *stop_monotonic_ut) {
93
+ if (cancelled && __atomic_load_n(cancelled, __ATOMIC_RELAXED)) {
94
+ internal_error(true, "Function has been cancelled");
95
+ return ND_SD_JOURNAL_CANCELLED;
96
+ }
97
+
98
+ if (now_monotonic_usec() > __atomic_load_n(stop_monotonic_ut, __ATOMIC_RELAXED)) {
99
+ internal_error(true, "Function timed out");
100
+ return ND_SD_JOURNAL_TIMED_OUT;
101
+ }
102
+
103
+ return ND_SD_JOURNAL_OK;
104
+}
105
+
106
+static __always_inline
107
+ND_SD_JOURNAL_STATUS nd_sd_journal_query_backward(
108
+ NsdJournal *j,
109
+ BUFFER *wb __maybe_unused,
110
+ FACETS *facets,
111
+ struct nd_journal_file *njf,
112
+ LOGS_QUERY_STATUS *fqs)
113
+{
114
+ usec_t anchor_delta = __atomic_load_n(&njf->max_journal_vs_realtime_delta_ut, __ATOMIC_RELAXED);
115
+ lqs_query_timeframe(fqs, anchor_delta);
116
+ usec_t start_ut = fqs->query.start_ut;
117
+ usec_t stop_ut = fqs->query.stop_ut;
118
+ bool stop_when_full = fqs->query.stop_when_full;
119
+
120
+ fqs->c.query_file.start_ut = start_ut;
121
+ fqs->c.query_file.stop_ut = stop_ut;
122
+
123
+ if (!nd_sd_journal_seek_to(j, start_ut))
124
+ return ND_SD_JOURNAL_FAILED_TO_SEEK;
125
+
126
+ size_t errors_no_timestamp = 0;
127
+ usec_t latest_msg_ut = 0; // the biggest timestamp we have seen so far
128
+ usec_t first_msg_ut = 0; // the first message we got from the db
129
+ size_t row_counter = 0, last_row_counter = 0, rows_useful = 0;
130
+ size_t bytes = 0, last_bytes = 0;
131
+
132
+ usec_t last_usec_from = 0;
133
+ usec_t last_usec_to = 0;
134
+
135
+ ND_SD_JOURNAL_STATUS status = ND_SD_JOURNAL_OK;
136
+
137
+ facets_rows_begin(facets);
138
+ while (status == ND_SD_JOURNAL_OK && nsd_journal_previous(j) > 0) {
139
+ usec_t msg_ut = 0;
140
+ if (nsd_journal_get_realtime_usec(j, &msg_ut) < 0 || !msg_ut) {
141
+ errors_no_timestamp++;
142
+ continue;
143
+ }
144
+
145
+ if (unlikely(msg_ut > start_ut))
146
+ continue;
147
+
148
+ if (unlikely(msg_ut < stop_ut))
149
+ break;
150
+
151
+ if (unlikely(msg_ut > latest_msg_ut))
152
+ latest_msg_ut = msg_ut;
153
+
154
+ if (unlikely(!first_msg_ut)) {
155
+ first_msg_ut = msg_ut;
156
+ fqs->c.query_file.first_msg_ut = msg_ut;
157
+
158
+#ifdef HAVE_SD_JOURNAL_GET_SEQNUM
159
+ if (nsd_journal_get_seqnum(j, &fqs->c.query_file.first_msg_seqnum, &fqs->c.query_file.first_msg_writer) <
160
+ 0) {
161
+ fqs->c.query_file.first_msg_seqnum = 0;
162
+ fqs->c.query_file.first_msg_writer = NSD_ID128_NULL;
163
+ }
164
+#endif
165
+ }
166
+
167
+ sampling_t sample = is_row_in_sample(
168
+ j, fqs, njf, msg_ut, FACETS_ANCHOR_DIRECTION_BACKWARD, facets_row_candidate_to_keep(facets, msg_ut));
169
+
170
+ if (sample == SAMPLING_FULL) {
171
+ bytes += nd_sd_journal_process_row(j, facets, njf, &msg_ut);
172
+
173
+ // make sure each line gets a unique timestamp
174
+ if (unlikely(msg_ut >= last_usec_from && msg_ut <= last_usec_to))
175
+ msg_ut = --last_usec_from;
176
+ else
177
+ last_usec_from = last_usec_to = msg_ut;
178
+
179
+ if (facets_row_finished(facets, msg_ut))
180
+ rows_useful++;
181
+
182
+ row_counter++;
183
+ if (unlikely(
184
+ (row_counter % FUNCTION_DATA_ONLY_CHECK_EVERY_ROWS) == 0 && stop_when_full &&
185
+ facets_rows(facets) >= fqs->rq.entries)) {
186
+ // stop the data only query
187
+ usec_t oldest = facets_row_oldest_ut(facets);
188
+ if (oldest && msg_ut < (oldest - anchor_delta))
189
+ break;
190
+ }
191
+
192
+ if (unlikely(row_counter % FUNCTION_PROGRESS_EVERY_ROWS == 0)) {
193
+ FUNCTION_PROGRESS_UPDATE_ROWS(fqs->c.rows_read, row_counter - last_row_counter);
194
+ last_row_counter = row_counter;
195
+
196
+ FUNCTION_PROGRESS_UPDATE_BYTES(fqs->c.bytes_read, bytes - last_bytes);
197
+ last_bytes = bytes;
198
+
199
+ status = check_stop(fqs->cancelled, fqs->stop_monotonic_ut);
200
+ }
201
+ } else if (sample == SAMPLING_SKIP_FIELDS)
202
+ facets_row_finished_unsampled(facets, msg_ut);
203
+ else {
204
+ sampling_update_running_query_file_estimates(facets, j, fqs, njf, msg_ut, FACETS_ANCHOR_DIRECTION_BACKWARD);
205
+ break;
206
+ }
207
+ }
208
+
209
+ FUNCTION_PROGRESS_UPDATE_ROWS(fqs->c.rows_read, row_counter - last_row_counter);
210
+ FUNCTION_PROGRESS_UPDATE_BYTES(fqs->c.bytes_read, bytes - last_bytes);
211
+
212
+ fqs->c.rows_useful += rows_useful;
213
+
214
+ if (errors_no_timestamp)
215
+ netdata_log_error("SYSTEMD-JOURNAL: %zu lines did not have timestamps", errors_no_timestamp);
216
+
217
+ if (latest_msg_ut > fqs->last_modified)
218
+ fqs->last_modified = latest_msg_ut;
219
+
220
+ return status;
221
+}
222
+
223
+static __always_inline
224
+ND_SD_JOURNAL_STATUS nd_sd_journal_query_forward(
225
+ NsdJournal *j,
226
+ BUFFER *wb __maybe_unused,
227
+ FACETS *facets,
228
+ struct nd_journal_file *njf,
229
+ LOGS_QUERY_STATUS *fqs)
230
+{
231
+ usec_t anchor_delta = __atomic_load_n(&njf->max_journal_vs_realtime_delta_ut, __ATOMIC_RELAXED);
232
+ lqs_query_timeframe(fqs, anchor_delta);
233
+ usec_t start_ut = fqs->query.start_ut;
234
+ usec_t stop_ut = fqs->query.stop_ut;
235
+ bool stop_when_full = fqs->query.stop_when_full;
236
+
237
+ fqs->c.query_file.start_ut = start_ut;
238
+ fqs->c.query_file.stop_ut = stop_ut;
239
+
240
+ if (!nd_sd_journal_seek_to(j, start_ut))
241
+ return ND_SD_JOURNAL_FAILED_TO_SEEK;
242
+
243
+ size_t errors_no_timestamp = 0;
244
+ usec_t latest_msg_ut = 0; // the biggest timestamp we have seen so far
245
+ usec_t first_msg_ut = 0; // the first message we got from the db
246
+ size_t row_counter = 0, last_row_counter = 0, rows_useful = 0;
247
+ size_t bytes = 0, last_bytes = 0;
248
+
249
+ usec_t last_usec_from = 0;
250
+ usec_t last_usec_to = 0;
251
+
252
+ ND_SD_JOURNAL_STATUS status = ND_SD_JOURNAL_OK;
253
+
254
+ facets_rows_begin(facets);
255
+ while (status == ND_SD_JOURNAL_OK && nsd_journal_next(j) > 0) {
256
+ usec_t msg_ut = 0;
257
+ if (nsd_journal_get_realtime_usec(j, &msg_ut) < 0 || !msg_ut) {
258
+ errors_no_timestamp++;
259
+ continue;
260
+ }
261
+
262
+ if (unlikely(msg_ut < start_ut))
263
+ continue;
264
+
265
+ if (unlikely(msg_ut > stop_ut))
266
+ break;
267
+
268
+ if (likely(msg_ut > latest_msg_ut))
269
+ latest_msg_ut = msg_ut;
270
+
271
+ if (unlikely(!first_msg_ut)) {
272
+ first_msg_ut = msg_ut;
273
+ fqs->c.query_file.first_msg_ut = msg_ut;
274
+ }
275
+
276
+ sampling_t sample = is_row_in_sample(
277
+ j, fqs, njf, msg_ut, FACETS_ANCHOR_DIRECTION_FORWARD, facets_row_candidate_to_keep(facets, msg_ut));
278
+
279
+ if (sample == SAMPLING_FULL) {
280
+ bytes += nd_sd_journal_process_row(j, facets, njf, &msg_ut);
281
+
282
+ // make sure each line gets a unique timestamp
283
+ if (unlikely(msg_ut >= last_usec_from && msg_ut <= last_usec_to))
284
+ msg_ut = ++last_usec_to;
285
+ else
286
+ last_usec_from = last_usec_to = msg_ut;
287
+
288
+ if (facets_row_finished(facets, msg_ut))
289
+ rows_useful++;
290
+
291
+ row_counter++;
292
+ if (unlikely(
293
+ (row_counter % FUNCTION_DATA_ONLY_CHECK_EVERY_ROWS) == 0 && stop_when_full &&
294
+ facets_rows(facets) >= fqs->rq.entries)) {
295
+ // stop the data only query
296
+ usec_t newest = facets_row_newest_ut(facets);
297
+ if (newest && msg_ut > (newest + anchor_delta))
298
+ break;
299
+ }
300
+
301
+ if (unlikely(row_counter % FUNCTION_PROGRESS_EVERY_ROWS == 0)) {
302
+ FUNCTION_PROGRESS_UPDATE_ROWS(fqs->c.rows_read, row_counter - last_row_counter);
303
+ last_row_counter = row_counter;
304
+
305
+ FUNCTION_PROGRESS_UPDATE_BYTES(fqs->c.bytes_read, bytes - last_bytes);
306
+ last_bytes = bytes;
307
+
308
+ status = check_stop(fqs->cancelled, fqs->stop_monotonic_ut);
309
+ }
310
+ } else if (sample == SAMPLING_SKIP_FIELDS)
311
+ facets_row_finished_unsampled(facets, msg_ut);
312
+ else {
313
+ sampling_update_running_query_file_estimates(facets, j, fqs, njf, msg_ut, FACETS_ANCHOR_DIRECTION_FORWARD);
314
+ break;
315
+ }
316
+ }
317
+
318
+ FUNCTION_PROGRESS_UPDATE_ROWS(fqs->c.rows_read, row_counter - last_row_counter);
319
+ FUNCTION_PROGRESS_UPDATE_BYTES(fqs->c.bytes_read, bytes - last_bytes);
320
+
321
+ fqs->c.rows_useful += rows_useful;
322
+
323
+ if (errors_no_timestamp)
324
+ netdata_log_error("SYSTEMD-JOURNAL: %zu lines did not have timestamps", errors_no_timestamp);
325
+
326
+ if (latest_msg_ut > fqs->last_modified)
327
+ fqs->last_modified = latest_msg_ut;
328
+
329
+ return status;
330
+}
331
+
332
+bool nd_sd_journal_check_if_modified_since(NsdJournal *j, usec_t seek_to, usec_t last_modified)
333
+{
334
+ // return true, if data have been modified since the timestamp
335
+
336
+ if (!last_modified || !seek_to)
337
+ return false;
338
+
339
+ if (!nd_sd_journal_seek_to(j, seek_to))
340
+ return false;
341
+
342
+ usec_t first_msg_ut = 0;
343
+ while (nsd_journal_previous(j) > 0) {
344
+ usec_t msg_ut;
345
+ if (nsd_journal_get_realtime_usec(j, &msg_ut) < 0)
346
+ continue;
347
+
348
+ first_msg_ut = msg_ut;
349
+ break;
350
+ }
351
+
352
+ return first_msg_ut != last_modified;
353
+}
354
+
355
+#ifdef HAVE_SD_JOURNAL_RESTART_FIELDS
356
+static __always_inline
357
+bool netdata_systemd_filtering_by_journal(NsdJournal *j, FACETS *facets, LOGS_QUERY_STATUS *lqs) {
358
+ const char *field = NULL;
359
+ const void *data = NULL;
360
+ size_t data_length;
361
+ size_t added_keys = 0;
362
+ size_t failures = 0;
363
+ size_t filters_added = 0;
364
+
365
+ NSD_JOURNAL_FOREACH_FIELD(j, field)
366
+ { // for each key
367
+ bool interesting;
368
+
369
+ if (lqs->rq.data_only)
370
+ interesting = facets_key_name_is_filter(facets, field);
371
+ else
372
+ interesting = facets_key_name_is_facet(facets, field);
373
+
374
+ if (interesting) {
375
+ int r = nsd_journal_query_unique(j, field);
376
+ if (r >= 0) {
377
+ bool added_this_key = false;
378
+ size_t added_values = 0;
379
+
380
+ nsd_journal_restart_unique(j);
381
+ while ((r = nsd_journal_enumerate_available_unique(j, &data, &data_length)) > 0) {
382
+ // for each value of the key
383
+ const char *key, *value;
384
+ size_t key_length, value_length;
385
+
386
+ if (!parse_journal_field(data, data_length, &key, &key_length, &value, &value_length))
387
+ continue;
388
+
389
+ facets_add_possible_value_name_to_key(facets, key, key_length, value, value_length);
390
+
391
+ if (!facets_key_name_value_length_is_selected(facets, key, key_length, value, value_length))
392
+ continue;
393
+
394
+ if (added_keys && !added_this_key) {
395
+ if (nsd_journal_add_conjunction(j) < 0) // key AND key AND key
396
+ failures++;
397
+
398
+ added_this_key = true;
399
+ added_keys++;
400
+ } else if (added_values)
401
+ if (nsd_journal_add_disjunction(j) < 0) // value OR value OR value
402
+ failures++;
403
+
404
+ if (nsd_journal_add_match(j, data, data_length) < 0)
405
+ failures++;
406
+
407
+ if (!added_keys) {
408
+ added_keys++;
409
+ added_this_key = true;
410
+ }
411
+
412
+ added_values++;
413
+ filters_added++;
414
+ }
415
+
416
+ if (r < 0)
417
+ failures++;
418
+ } else {
419
+ failures++;
420
+ }
421
+ }
422
+ }
423
+
424
+ if (failures) {
425
+ lqs_log_error(lqs, "failed to setup journal filter, will run the full query.");
426
+ nsd_journal_flush_matches(j);
427
+ return true;
428
+ }
429
+
430
+ return filters_added ? true : false;
431
+}
432
+#endif // HAVE_SD_JOURNAL_RESTART_FIELDS
433
+
434
+static __always_inline
435
+ND_SD_JOURNAL_STATUS nd_sd_journal_query_one_file(
436
+ const char *filename,
437
+ BUFFER *wb,
438
+ FACETS *facets,
439
+ struct nd_journal_file *njf,
440
+ LOGS_QUERY_STATUS *fqs)
441
+{
442
+ NsdJournal *j = NULL;
443
+ errno_clear();
444
+
445
+ fstat_cache_enable_on_thread();
446
+
447
+ const char *paths[2] = {
448
+ [0] = filename,
449
+ [1] = NULL,
450
+ };
451
+
452
+ if (nsd_journal_open_files(&j, paths, ND_SD_JOURNAL_OPEN_FLAGS) < 0 || !j) {
453
+ netdata_log_error("JOURNAL: cannot open file '%s' for query", filename);
454
+ fstat_cache_disable_on_thread();
455
+ return ND_SD_JOURNAL_FAILED_TO_OPEN;
456
+ }
457
+
458
+ ND_SD_JOURNAL_STATUS status;
459
+ bool matches_filters = true;
460
+
461
+#ifdef HAVE_SD_JOURNAL_RESTART_FIELDS
462
+ if (fqs->rq.slice) {
463
+ usec_t started = now_monotonic_usec();
464
+
465
+ matches_filters = netdata_systemd_filtering_by_journal(j, facets, fqs) || !fqs->rq.filters;
466
+ usec_t ended = now_monotonic_usec();
467
+
468
+ fqs->c.matches_setup_ut += (ended - started);
469
+ }
470
+#endif // HAVE_SD_JOURNAL_RESTART_FIELDS
471
+
472
+ if (matches_filters) {
473
+ if (fqs->rq.direction == FACETS_ANCHOR_DIRECTION_FORWARD)
474
+ status = nd_sd_journal_query_forward(j, wb, facets, njf, fqs);
475
+ else
476
+ status = nd_sd_journal_query_backward(j, wb, facets, njf, fqs);
477
+ } else
478
+ status = ND_SD_JOURNAL_NO_FILE_MATCHED;
479
+
480
+ nsd_journal_close(j);
481
+ fstat_cache_disable_on_thread();
482
+
483
+ return status;
484
+}
485
+
486
+static __always_inline
487
+bool jf_is_mine(struct nd_journal_file *njf, LOGS_QUERY_STATUS *fqs) {
488
+ if ((fqs->rq.source_type == ND_SD_JF_NONE && !fqs->rq.sources) || (njf->source_type & fqs->rq.source_type) ||
489
+ (fqs->rq.sources && simple_pattern_matches(fqs->rq.sources, string2str(njf->source)))) {
490
+ if (!njf->msg_last_ut)
491
+ // the file is not scanned yet, or the timestamps have not been updated,
492
+ // so we don't know if it can contribute or not - let's add it.
493
+ return true;
494
+
495
+ usec_t anchor_delta = JOURNAL_VS_REALTIME_DELTA_MAX_UT;
496
+ usec_t first_ut = njf->msg_first_ut - anchor_delta;
497
+ usec_t last_ut = njf->msg_last_ut + anchor_delta;
498
+
499
+ if (last_ut >= fqs->rq.after_ut && first_ut <= fqs->rq.before_ut)
500
+ return true;
501
+ }
502
+
503
+ return false;
504
+}
505
+
506
+static __always_inline
507
+int nd_sd_journal_query(BUFFER *wb, LOGS_QUERY_STATUS *lqs) {
508
+ FACETS *facets = lqs->facets;
509
+
510
+ ND_SD_JOURNAL_STATUS status = ND_SD_JOURNAL_NO_FILE_MATCHED;
511
+ struct nd_journal_file *njf;
512
+
513
+ lqs->c.files_matched = 0;
514
+ lqs->c.file_working = 0;
515
+ lqs->c.rows_useful = 0;
516
+ lqs->c.rows_read = 0;
517
+ lqs->c.bytes_read = 0;
518
+
519
+ size_t files_used = 0;
520
+ size_t files_max = dictionary_entries(nd_journal_files_registry);
521
+ const DICTIONARY_ITEM *file_items[files_max];
522
+
523
+ // count the files
524
+ bool files_are_newer = false;
525
+ dfe_start_read(nd_journal_files_registry, njf)
526
+ {
527
+ if (!jf_is_mine(njf, lqs))
528
+ continue;
529
+
530
+ file_items[files_used++] = dictionary_acquired_item_dup(nd_journal_files_registry, njf_dfe.item);
531
+
532
+ if (njf->msg_last_ut > lqs->rq.if_modified_since)
533
+ files_are_newer = true;
534
+ }
535
+ dfe_done(jf);
536
+
537
+ lqs->c.files_matched = files_used;
538
+
539
+ if (lqs->rq.if_modified_since && !files_are_newer) {
540
+ // release the files
541
+ for (size_t f = 0; f < files_used; f++)
542
+ dictionary_acquired_item_release(nd_journal_files_registry, file_items[f]);
543
+
544
+ return rrd_call_function_error(wb, "No new data since the previous call.", HTTP_RESP_NOT_MODIFIED);
545
+ }
546
+
547
+ // sort the files, so that they are optimal for facets
548
+ if (files_used >= 2) {
549
+ if (lqs->rq.direction == FACETS_ANCHOR_DIRECTION_BACKWARD)
550
+ qsort(file_items, files_used, sizeof(const DICTIONARY_ITEM *), nd_journal_file_dict_items_backward_compar);
551
+ else
552
+ qsort(file_items, files_used, sizeof(const DICTIONARY_ITEM *), nd_journal_file_dict_items_forward_compar);
553
+ }
554
+
555
+ bool partial = false;
556
+ usec_t query_started_ut = now_monotonic_usec();
557
+ usec_t started_ut = query_started_ut;
558
+ usec_t ended_ut = started_ut;
559
+ usec_t duration_ut = 0, max_duration_ut = 0;
560
+ usec_t progress_duration_ut = 0;
561
+
562
+ sampling_query_init(lqs, facets);
563
+
564
+ buffer_json_member_add_array(wb, "_journal_files");
565
+ for (size_t f = 0; f < files_used; f++) {
566
+ const char *filename = dictionary_acquired_item_name(file_items[f]);
567
+ njf = dictionary_acquired_item_value(file_items[f]);
568
+
569
+ if (!jf_is_mine(njf, lqs))
570
+ continue;
571
+
572
+ started_ut = ended_ut;
573
+
574
+ // do not even try to do the query if we expect it to pass the timeout
575
+ if (ended_ut + max_duration_ut * 3 >= *lqs->stop_monotonic_ut) {
576
+ partial = true;
577
+ status = ND_SD_JOURNAL_TIMED_OUT;
578
+ break;
579
+ }
580
+
581
+ lqs->c.file_working++;
582
+
583
+ size_t fs_calls = fstat_thread_calls;
584
+ size_t fs_cached = fstat_thread_cached_responses;
585
+ size_t rows_useful = lqs->c.rows_useful;
586
+ size_t rows_read = lqs->c.rows_read;
587
+ size_t bytes_read = lqs->c.bytes_read;
588
+ size_t matches_setup_ut = lqs->c.matches_setup_ut;
589
+
590
+ sampling_file_init(lqs, njf);
591
+
592
+ ND_SD_JOURNAL_STATUS tmp_status = nd_sd_journal_query_one_file(filename, wb, facets, njf, lqs);
593
+
594
+ rows_useful = lqs->c.rows_useful - rows_useful;
595
+ rows_read = lqs->c.rows_read - rows_read;
596
+ bytes_read = lqs->c.bytes_read - bytes_read;
597
+ matches_setup_ut = lqs->c.matches_setup_ut - matches_setup_ut;
598
+ fs_calls = fstat_thread_calls - fs_calls;
599
+ fs_cached = fstat_thread_cached_responses - fs_cached;
600
+
601
+ ended_ut = now_monotonic_usec();
602
+ duration_ut = ended_ut - started_ut;
603
+
604
+ if (duration_ut > max_duration_ut)
605
+ max_duration_ut = duration_ut;
606
+
607
+ progress_duration_ut += duration_ut;
608
+ if (progress_duration_ut >= ND_SD_JOURNAL_PROGRESS_EVERY_UT) {
609
+ progress_duration_ut = 0;
610
+ netdata_mutex_lock(&stdout_mutex);
611
+ pluginsd_function_progress_to_stdout(lqs->rq.transaction, f + 1, files_used);
612
+ netdata_mutex_unlock(&stdout_mutex);
613
+ }
614
+
615
+ buffer_json_add_array_item_object(wb); // journal file
616
+ {
617
+ // information about the file
618
+ buffer_json_member_add_string(wb, "_filename", filename);
619
+ buffer_json_member_add_uint64(wb, "_source_type", njf->source_type);
620
+ buffer_json_member_add_string(wb, "_source", string2str(njf->source));
621
+ buffer_json_member_add_uint64(wb, "_last_modified_ut", njf->file_last_modified_ut);
622
+ buffer_json_member_add_uint64(wb, "_msg_first_ut", njf->msg_first_ut);
623
+ buffer_json_member_add_uint64(wb, "_msg_last_ut", njf->msg_last_ut);
624
+ buffer_json_member_add_uint64(wb, "_journal_vs_realtime_delta_ut", njf->max_journal_vs_realtime_delta_ut);
625
+
626
+ // information about the current use of the file
627
+ buffer_json_member_add_uint64(wb, "duration_ut", ended_ut - started_ut);
628
+ buffer_json_member_add_uint64(wb, "rows_read", rows_read);
629
+ buffer_json_member_add_uint64(wb, "rows_useful", rows_useful);
630
+ buffer_json_member_add_double(
631
+ wb, "rows_per_second", (double)rows_read / (double)duration_ut * (double)USEC_PER_SEC);
632
+ buffer_json_member_add_uint64(wb, "bytes_read", bytes_read);
633
+ buffer_json_member_add_double(
634
+ wb, "bytes_per_second", (double)bytes_read / (double)duration_ut * (double)USEC_PER_SEC);
635
+ buffer_json_member_add_uint64(wb, "duration_matches_ut", matches_setup_ut);
636
+ buffer_json_member_add_uint64(wb, "fstat_query_calls", fs_calls);
637
+ buffer_json_member_add_uint64(wb, "fstat_query_cached_responses", fs_cached);
638
+
639
+ if (lqs->rq.sampling) {
640
+ buffer_json_member_add_object(wb, "_sampling");
641
+ {
642
+ buffer_json_member_add_uint64(wb, "sampled", lqs->c.samples_per_file.sampled);
643
+ buffer_json_member_add_uint64(wb, "unsampled", lqs->c.samples_per_file.unsampled);
644
+ buffer_json_member_add_uint64(wb, "estimated", lqs->c.samples_per_file.estimated);
645
+ }
646
+ buffer_json_object_close(wb); // _sampling
647
+ }
648
+ }
649
+ buffer_json_object_close(wb); // journal file
650
+
651
+ bool stop = false;
652
+ switch (tmp_status) {
653
+ case ND_SD_JOURNAL_OK:
654
+ case ND_SD_JOURNAL_NO_FILE_MATCHED:
655
+ status = (status == ND_SD_JOURNAL_OK) ? ND_SD_JOURNAL_OK : tmp_status;
656
+ break;
657
+
658
+ case ND_SD_JOURNAL_FAILED_TO_OPEN:
659
+ case ND_SD_JOURNAL_FAILED_TO_SEEK:
660
+ partial = true;
661
+ if (status == ND_SD_JOURNAL_NO_FILE_MATCHED)
662
+ status = tmp_status;
663
+ break;
664
+
665
+ case ND_SD_JOURNAL_CANCELLED:
666
+ case ND_SD_JOURNAL_TIMED_OUT:
667
+ partial = true;
668
+ stop = true;
669
+ status = tmp_status;
670
+ break;
671
+
672
+ case ND_SD_JOURNAL_NOT_MODIFIED:
673
+ internal_fatal(true, "this should never be returned here");
674
+ break;
675
+ }
676
+
677
+ if (stop)
678
+ break;
679
+ }
680
+ buffer_json_array_close(wb); // _journal_files
681
+
682
+ // release the files
683
+ for (size_t f = 0; f < files_used; f++)
684
+ dictionary_acquired_item_release(nd_journal_files_registry, file_items[f]);
685
+
686
+ switch (status) {
687
+ case ND_SD_JOURNAL_OK:
688
+ if (lqs->rq.if_modified_since && !lqs->c.rows_useful)
689
+ return rrd_call_function_error(
690
+ wb, "No additional useful data since the previous call.", HTTP_RESP_NOT_MODIFIED);
691
+ break;
692
+
693
+ case ND_SD_JOURNAL_TIMED_OUT:
694
+ case ND_SD_JOURNAL_NO_FILE_MATCHED:
695
+ break;
696
+
697
+ case ND_SD_JOURNAL_CANCELLED:
698
+ return rrd_call_function_error(wb, "Request cancelled.", HTTP_RESP_CLIENT_CLOSED_REQUEST);
699
+
700
+ case ND_SD_JOURNAL_NOT_MODIFIED:
701
+ return rrd_call_function_error(wb, "No new data since the previous call.", HTTP_RESP_NOT_MODIFIED);
702
+
703
+ case ND_SD_JOURNAL_FAILED_TO_OPEN:
704
+ return rrd_call_function_error(wb, "Failed to open systemd journal file.", HTTP_RESP_INTERNAL_SERVER_ERROR);
705
+
706
+ case ND_SD_JOURNAL_FAILED_TO_SEEK:
707
+ return rrd_call_function_error(
708
+ wb, "Failed to seek in systemd journal file.", HTTP_RESP_INTERNAL_SERVER_ERROR);
709
+
710
+ default:
711
+ return rrd_call_function_error(wb, "Unknown status", HTTP_RESP_INTERNAL_SERVER_ERROR);
712
+ }
713
+
714
+ buffer_json_member_add_uint64(wb, "status", HTTP_RESP_OK);
715
+ buffer_json_member_add_boolean(wb, "partial", partial);
716
+ buffer_json_member_add_string(wb, "type", "table");
717
+
718
+ // build a message for the query
719
+ if (!lqs->rq.data_only) {
720
+ CLEAN_BUFFER *msg = buffer_create(0, NULL);
721
+ CLEAN_BUFFER *msg_description = buffer_create(0, NULL);
722
+ ND_LOG_FIELD_PRIORITY msg_priority = NDLP_INFO;
723
+
724
+ if (!nd_journal_files_completed_once()) {
725
+ buffer_strcat(msg, "Journals are still being scanned. ");
726
+ buffer_strcat(
727
+ msg_description,
728
+ "LIBRARY SCAN: The journal files are still being scanned, you are probably viewing incomplete data. ");
729
+ msg_priority = NDLP_WARNING;
730
+ }
731
+
732
+ if (partial) {
733
+ buffer_strcat(msg, "Query timed-out, incomplete data. ");
734
+ buffer_strcat(
735
+ msg_description,
736
+ "QUERY TIMEOUT: The query timed out and may not include all the data of the selected window. ");
737
+ msg_priority = NDLP_WARNING;
738
+ }
739
+
740
+ if (lqs->c.samples.estimated || lqs->c.samples.unsampled) {
741
+ double percent = (double)(lqs->c.samples.sampled * 100.0 /
742
+ (lqs->c.samples.estimated + lqs->c.samples.unsampled + lqs->c.samples.sampled));
743
+ buffer_sprintf(msg, "%.2f%% real data", percent);
744
+ buffer_sprintf(msg_description, "ACTUAL DATA: The filters counters reflect %0.2f%% of the data. ", percent);
745
+ msg_priority = MIN(msg_priority, NDLP_NOTICE);
746
+ }
747
+
748
+ if (lqs->c.samples.unsampled) {
749
+ double percent = (double)(lqs->c.samples.unsampled * 100.0 /
750
+ (lqs->c.samples.estimated + lqs->c.samples.unsampled + lqs->c.samples.sampled));
751
+ buffer_sprintf(msg, ", %.2f%% unsampled", percent);
752
+ buffer_sprintf(
753
+ msg_description,
754
+ "UNSAMPLED DATA: %0.2f%% of the events exist and have been counted, but their values have not been evaluated, so they are not included in the filters counters. ",
755
+ percent);
756
+ msg_priority = MIN(msg_priority, NDLP_NOTICE);
757
+ }
758
+
759
+ if (lqs->c.samples.estimated) {
760
+ double percent = (double)(lqs->c.samples.estimated * 100.0 /
761
+ (lqs->c.samples.estimated + lqs->c.samples.unsampled + lqs->c.samples.sampled));
762
+ buffer_sprintf(msg, ", %.2f%% estimated", percent);
763
+ buffer_sprintf(
764
+ msg_description,
765
+ "ESTIMATED DATA: The query selected a large amount of data, so to avoid delaying too much, the presented data are estimated by %0.2f%%. ",
766
+ percent);
767
+ msg_priority = MIN(msg_priority, NDLP_NOTICE);
768
+ }
769
+
770
+ buffer_json_member_add_object(wb, "message");
771
+ if (buffer_tostring(msg)) {
772
+ buffer_json_member_add_string(wb, "title", buffer_tostring(msg));
773
+ buffer_json_member_add_string(wb, "description", buffer_tostring(msg_description));
774
+ buffer_json_member_add_string(wb, "status", nd_log_id2priority(msg_priority));
775
+ }
776
+ // else send an empty object if there is nothing to tell
777
+ buffer_json_object_close(wb); // message
778
+ }
779
+
780
+ if (!lqs->rq.data_only) {
781
+ buffer_json_member_add_time_t(wb, "update_every", 1);
782
+ buffer_json_member_add_string(wb, "help", ND_SD_JOURNAL_FUNCTION_DESCRIPTION);
783
+ }
784
+
785
+ if (!lqs->rq.data_only || lqs->rq.tail)
786
+ buffer_json_member_add_uint64(wb, "last_modified", lqs->last_modified);
787
+
788
+ facets_sort_and_reorder_keys(facets);
789
+ facets_report(facets, wb, used_hashes_registry);
790
+
791
+ wb->expires = now_realtime_sec() + (lqs->rq.data_only ? 3600 : 0);
792
+ buffer_json_member_add_time_t(wb, "expires", wb->expires);
793
+
794
+ buffer_json_member_add_object(wb, "_fstat_caching");
795
+ {
796
+ buffer_json_member_add_uint64(wb, "calls", fstat_thread_calls);
797
+ buffer_json_member_add_uint64(wb, "cached", fstat_thread_cached_responses);
798
+ }
799
+ buffer_json_object_close(wb); // _fstat_caching
800
+
801
+ if (lqs->rq.sampling) {
802
+ buffer_json_member_add_object(wb, "_sampling");
803
+ {
804
+ buffer_json_member_add_uint64(wb, "sampled", lqs->c.samples.sampled);
805
+ buffer_json_member_add_uint64(wb, "unsampled", lqs->c.samples.unsampled);
806
+ buffer_json_member_add_uint64(wb, "estimated", lqs->c.samples.estimated);
807
+ }
808
+ buffer_json_object_close(wb); // _sampling
809
+ }
810
+
811
+ wb->content_type = CT_APPLICATION_JSON;
812
+ wb->response_code = HTTP_RESP_OK;
813
+ return wb->response_code;
814
+}
815
+
816
+#endif //NETDATA_SYSTEMD_JOURNAL_EXECUTE_H
src/collectors/systemd-journal.plugin/systemd-journal-function.h
new
+67
@@ -0,0 +1,67 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#ifndef NETDATA_SYSTEMD_JOURNAL_FUNCTION_H
4
+#define NETDATA_SYSTEMD_JOURNAL_FUNCTION_H
5
+
6
+#define ND_SD_JOURNAL_SAMPLING_SLOTS 1000
7
+#define ND_SD_JOURNAL_SAMPLING_RECALIBRATE 10000
8
+
9
+#define FACET_MAX_VALUE_LENGTH 8192
10
+#define ND_SD_JOURNAL_DEFAULT_TIMEOUT 60
11
+#define ND_SD_JOURNAL_PROGRESS_EVERY_UT (250 * USEC_PER_MS)
12
+
13
+#define JOURNAL_DEFAULT_DIRECTION FACETS_ANCHOR_DIRECTION_BACKWARD
14
+
15
+#define JD_SOURCE_REALTIME_TIMESTAMP "_SOURCE_REALTIME_TIMESTAMP"
16
+
17
+// structures needed by LQS
18
+struct lqs_extension {
19
+ struct {
20
+ usec_t start_ut;
21
+ usec_t stop_ut;
22
+ usec_t first_msg_ut;
23
+
24
+ NsdId128 first_msg_writer;
25
+ uint64_t first_msg_seqnum;
26
+ } query_file;
27
+
28
+ struct {
29
+ uint32_t enable_after_samples;
30
+ uint32_t slots;
31
+ uint32_t sampled;
32
+ uint32_t unsampled;
33
+ uint32_t estimated;
34
+ } samples;
35
+
36
+ struct {
37
+ uint32_t enable_after_samples;
38
+ uint32_t every;
39
+ uint32_t skipped;
40
+ uint32_t recalibrate;
41
+ uint32_t sampled;
42
+ uint32_t unsampled;
43
+ uint32_t estimated;
44
+ } samples_per_file;
45
+
46
+ struct {
47
+ usec_t start_ut;
48
+ usec_t end_ut;
49
+ usec_t step_ut;
50
+ uint32_t enable_after_samples;
51
+ uint32_t sampled[ND_SD_JOURNAL_SAMPLING_SLOTS];
52
+ uint32_t unsampled[ND_SD_JOURNAL_SAMPLING_SLOTS];
53
+ } samples_per_time_slot;
54
+
55
+ // per file progress info
56
+ // size_t cached_count;
57
+
58
+ // progress statistics
59
+ usec_t matches_setup_ut;
60
+ size_t rows_useful;
61
+ size_t rows_read;
62
+ size_t bytes_read;
63
+ size_t files_matched;
64
+ size_t file_working;
65
+};
66
+
67
+#endif //NETDATA_SYSTEMD_JOURNAL_FUNCTION_H
src/collectors/systemd-journal.plugin/systemd-journal-sampling.h
+22
-11
@@ -6,7 +6,8 @@
6
// ----------------------------------------------------------------------------
7
// sampling support
8
9
-static inline void sampling_query_init(LOGS_QUERY_STATUS *lqs, FACETS *facets)
9
+static __always_inline
10
+void sampling_query_init(LOGS_QUERY_STATUS *lqs, FACETS *facets)
11
{
12
if (!lqs->rq.sampling)
13
return;
@@ -71,7 +72,8 @@ static inline void sampling_query_init(LOGS_QUERY_STATUS *lqs, FACETS *facets)
72
lqs->c.samples_per_time_slot.enable_after_samples = lqs->rq.entries;
73
}
74
74
-static inline void sampling_file_init(LOGS_QUERY_STATUS *lqs, struct nd_journal_file *jf __maybe_unused)
75
+static __always_inline
76
+void sampling_file_init(LOGS_QUERY_STATUS *lqs, struct nd_journal_file *jf __maybe_unused)
77
{
78
lqs->c.samples_per_file.sampled = 0;
79
lqs->c.samples_per_file.unsampled = 0;
@@ -81,7 +83,8 @@ static inline void sampling_file_init(LOGS_QUERY_STATUS *lqs, struct nd_journal_
83
lqs->c.samples_per_file.recalibrate = 0;
84
}
85
84
-static inline size_t sampling_file_lines_scanned_so_far(LOGS_QUERY_STATUS *lqs)
86
+static __always_inline
87
+size_t sampling_file_lines_scanned_so_far(LOGS_QUERY_STATUS *lqs)
88
{
89
size_t sampled = lqs->c.samples_per_file.sampled + lqs->c.samples_per_file.unsampled;
90
if (!sampled)
@@ -89,7 +92,8 @@ static inline size_t sampling_file_lines_scanned_so_far(LOGS_QUERY_STATUS *lqs)
92
return sampled;
93
}
94
92
-static inline void sampling_running_file_query_overlapping_timeframe_ut(
95
+static __always_inline
96
+void sampling_running_file_query_overlapping_timeframe_ut(
97
LOGS_QUERY_STATUS *lqs,
98
struct nd_journal_file *jf,
99
FACETS_ANCHOR_DIRECTION direction,
@@ -135,7 +139,8 @@ static inline void sampling_running_file_query_overlapping_timeframe_ut(
139
*before_ut = newest_ut;
140
}
141
138
-static inline double sampling_running_file_query_progress_by_time(
142
+static __always_inline
143
+double sampling_running_file_query_progress_by_time(
144
LOGS_QUERY_STATUS *lqs,
145
struct nd_journal_file *jf,
146
FACETS_ANCHOR_DIRECTION direction,
@@ -155,7 +160,8 @@ static inline double sampling_running_file_query_progress_by_time(
160
return progress;
161
}
162
158
-static inline usec_t sampling_running_file_query_remaining_time(
163
+static __always_inline
164
+usec_t sampling_running_file_query_remaining_time(
165
LOGS_QUERY_STATUS *lqs,
166
struct nd_journal_file *jf,
167
FACETS_ANCHOR_DIRECTION direction,
@@ -199,7 +205,8 @@ static inline usec_t sampling_running_file_query_remaining_time(
205
return remaining_ut;
206
}
207
202
-static inline size_t sampling_running_file_query_estimate_remaining_lines_by_time(
208
+static __always_inline
209
+size_t sampling_running_file_query_estimate_remaining_lines_by_time(
210
LOGS_QUERY_STATUS *lqs,
211
struct nd_journal_file *jf,
212
FACETS_ANCHOR_DIRECTION direction,
@@ -254,7 +261,8 @@ static inline size_t sampling_running_file_query_estimate_remaining_lines_by_tim
261
return remaining_logs_by_time;
262
}
263
257
-static inline size_t sampling_running_file_query_estimate_remaining_lines(
264
+static __always_inline
265
+size_t sampling_running_file_query_estimate_remaining_lines(
266
NsdJournal *j __maybe_unused,
267
LOGS_QUERY_STATUS *lqs,
268
struct nd_journal_file *jf,
@@ -302,7 +310,8 @@ static inline size_t sampling_running_file_query_estimate_remaining_lines(
310
return sampling_running_file_query_estimate_remaining_lines_by_time(lqs, jf, direction, msg_ut);
311
}
312
305
-static inline void sampling_decide_file_sampling_every(
313
+static __always_inline
314
+void sampling_decide_file_sampling_every(
315
NsdJournal *j,
316
LOGS_QUERY_STATUS *lqs,
317
struct nd_journal_file *jf,
@@ -330,7 +339,8 @@ typedef enum {
339
SAMPLING_SKIP_FIELDS = 1,
340
} sampling_t;
341
333
-static inline sampling_t is_row_in_sample(
342
+static __always_inline
343
+sampling_t is_row_in_sample(
344
NsdJournal *j,
345
LOGS_QUERY_STATUS *lqs,
346
struct nd_journal_file *jf,
@@ -395,7 +405,8 @@ static inline sampling_t is_row_in_sample(
405
return SAMPLING_SKIP_FIELDS;
406
}
407
398
-static inline void sampling_update_running_query_file_estimates(
408
+static __always_inline
409
+void sampling_update_running_query_file_estimates(
410
FACETS *facets,
411
NsdJournal *j,
412
LOGS_QUERY_STATUS *lqs,
src/collectors/systemd-journal.plugin/systemd-journal.c
+30
-891
@@ -8,70 +8,33 @@
8
9
#include "collectors/systemd-journal.plugin/provider/netdata_provider.h"
10
#include "systemd-internals.h"
11
+#include "systemd-journal-function.h"
12
13
#define ND_SD_JOURNAL_FUNCTION_DESCRIPTION "View, search and analyze systemd journal entries."
14
#define ND_SD_JOURNAL_FUNCTION_NAME "systemd-journal"
14
-#define ND_SD_JOURNAL_SAMPLING_SLOTS 1000
15
-#define ND_SD_JOURNAL_SAMPLING_RECALIBRATE 10000
15
17
-#ifdef HAVE_SD_JOURNAL_RESTART_FIELDS
18
-#define LQS_DEFAULT_SLICE_MODE 1
19
-#else
20
-#define LQS_DEFAULT_SLICE_MODE 0
21
-#endif
16
+#define JOURNAL_KEY_ND_JOURNAL_PROCESS "ND_JOURNAL_PROCESS"
17
18
// functions needed by LQS
24
-static SD_JOURNAL_FILE_SOURCE_TYPE get_internal_source_type(const char *value);
25
-
26
-// structures needed by LQS
27
-struct lqs_extension {
28
- struct {
29
- usec_t start_ut;
30
- usec_t stop_ut;
31
- usec_t first_msg_ut;
32
-
33
- NsdId128 first_msg_writer;
34
- uint64_t first_msg_seqnum;
35
- } query_file;
36
-
37
- struct {
38
- uint32_t enable_after_samples;
39
- uint32_t slots;
40
- uint32_t sampled;
41
- uint32_t unsampled;
42
- uint32_t estimated;
43
- } samples;
44
-
45
- struct {
46
- uint32_t enable_after_samples;
47
- uint32_t every;
48
- uint32_t skipped;
49
- uint32_t recalibrate;
50
- uint32_t sampled;
51
- uint32_t unsampled;
52
- uint32_t estimated;
53
- } samples_per_file;
54
-
55
- struct {
56
- usec_t start_ut;
57
- usec_t end_ut;
58
- usec_t step_ut;
59
- uint32_t enable_after_samples;
60
- uint32_t sampled[ND_SD_JOURNAL_SAMPLING_SLOTS];
61
- uint32_t unsampled[ND_SD_JOURNAL_SAMPLING_SLOTS];
62
- } samples_per_time_slot;
63
-
64
- // per file progress info
65
- // size_t cached_count;
19
+static __always_inline
20
+SD_JOURNAL_FILE_SOURCE_TYPE get_internal_source_type(const char *value) {
21
+ if (strcmp(value, ND_SD_JF_SOURCE_ALL_NAME) == 0)
22
+ return ND_SD_JF_ALL;
23
+ else if (strcmp(value, ND_SD_JF_SOURCE_LOCAL_NAME) == 0)
24
+ return ND_SD_JF_LOCAL_ALL;
25
+ else if (strcmp(value, ND_SD_JF_SOURCE_REMOTES_NAME) == 0)
26
+ return ND_SD_JF_REMOTE_ALL;
27
+ else if (strcmp(value, ND_SD_JF_SOURCE_NAMESPACES_NAME) == 0)
28
+ return ND_SD_JF_LOCAL_NAMESPACE;
29
+ else if (strcmp(value, ND_SD_JF_SOURCE_LOCAL_SYSTEM_NAME) == 0)
30
+ return ND_SD_JF_LOCAL_SYSTEM;
31
+ else if (strcmp(value, ND_SD_JF_SOURCE_LOCAL_USERS_NAME) == 0)
32
+ return ND_SD_JF_LOCAL_USER;
33
+ else if (strcmp(value, ND_SD_JF_SOURCE_LOCAL_OTHER_NAME) == 0)
34
+ return ND_SD_JF_LOCAL_OTHER;
35
67
- // progress statistics
68
- usec_t matches_setup_ut;
69
- size_t rows_useful;
70
- size_t rows_read;
71
- size_t bytes_read;
72
- size_t files_matched;
73
- size_t file_working;
74
-};
36
+ return ND_SD_JF_NONE;
37
+}
38
39
// prepare LQS
40
#define LQS_FUNCTION_NAME ND_SD_JOURNAL_FUNCTION_NAME
@@ -84,16 +47,15 @@ struct lqs_extension {
47
#define LQS_PARAMETER_SOURCE_NAME "Journal Sources" // this is how it is shown to users
48
#define LQS_FUNCTION_GET_INTERNAL_SOURCE_TYPE(value) get_internal_source_type(value)
49
#define LQS_FUNCTION_SOURCE_TO_JSON_ARRAY(wb) available_journal_file_sources_to_json_array(wb)
87
-#include "libnetdata/facets/logs_query_status.h"
50
89
-#include "systemd-journal-sampling.h"
51
+#ifdef HAVE_SD_JOURNAL_RESTART_FIELDS
52
+#define LQS_DEFAULT_SLICE_MODE 1
53
+#else
54
+#define LQS_DEFAULT_SLICE_MODE 0
55
+#endif
56
+
57
+#include "libnetdata/facets/logs_query_status.h"
58
91
-#define FACET_MAX_VALUE_LENGTH 8192
92
-#define ND_SD_JOURNAL_DEFAULT_TIMEOUT 60
93
-#define ND_SD_JOURNAL_PROGRESS_EVERY_UT (250 * USEC_PER_MS)
94
-#define JOURNAL_KEY_ND_JOURNAL_FILE "ND_JOURNAL_FILE"
95
-#define JOURNAL_KEY_ND_JOURNAL_PROCESS "ND_JOURNAL_PROCESS"
96
-#define JOURNAL_DEFAULT_DIRECTION FACETS_ANCHOR_DIRECTION_BACKWARD
59
#define SYSTEMD_ALWAYS_VISIBLE_KEYS NULL
60
61
#define SYSTEMD_KEYS_EXCLUDED_FROM_FACETS \
@@ -191,833 +153,10 @@ struct lqs_extension {
153
\
154
""
155
194
-static SD_JOURNAL_FILE_SOURCE_TYPE get_internal_source_type(const char *value)
195
-{
196
- if (strcmp(value, ND_SD_JF_SOURCE_ALL_NAME) == 0)
197
- return ND_SD_JF_ALL;
198
- else if (strcmp(value, ND_SD_JF_SOURCE_LOCAL_NAME) == 0)
199
- return ND_SD_JF_LOCAL_ALL;
200
- else if (strcmp(value, ND_SD_JF_SOURCE_REMOTES_NAME) == 0)
201
- return ND_SD_JF_REMOTE_ALL;
202
- else if (strcmp(value, ND_SD_JF_SOURCE_NAMESPACES_NAME) == 0)
203
- return ND_SD_JF_LOCAL_NAMESPACE;
204
- else if (strcmp(value, ND_SD_JF_SOURCE_LOCAL_SYSTEM_NAME) == 0)
205
- return ND_SD_JF_LOCAL_SYSTEM;
206
- else if (strcmp(value, ND_SD_JF_SOURCE_LOCAL_USERS_NAME) == 0)
207
- return ND_SD_JF_LOCAL_USER;
208
- else if (strcmp(value, ND_SD_JF_SOURCE_LOCAL_OTHER_NAME) == 0)
209
- return ND_SD_JF_LOCAL_OTHER;
210
-
211
- return ND_SD_JF_NONE;
212
-}
213
-
214
-static inline bool nd_sd_journal_seek_to(NsdJournal *j, usec_t timestamp)
215
-{
216
- if (nsd_journal_seek_realtime_usec(j, timestamp) < 0) {
217
- netdata_log_error("SYSTEMD-JOURNAL: Failed to seek to %" PRIu64, timestamp);
218
- if (nsd_journal_seek_tail(j) < 0) {
219
- netdata_log_error("SYSTEMD-JOURNAL: Failed to seek to journal's tail");
220
- return false;
221
- }
222
- }
223
-
224
- return true;
225
-}
226
-
227
-#define JD_SOURCE_REALTIME_TIMESTAMP "_SOURCE_REALTIME_TIMESTAMP"
228
-
229
-static inline size_t
230
-nd_sd_journal_process_row(NsdJournal *j, FACETS *facets, struct nd_journal_file *njf, usec_t *msg_ut)
231
-{
232
- const void *data;
233
- size_t length, bytes = 0;
234
-
235
- facets_add_key_value_length(
236
- facets, JOURNAL_KEY_ND_JOURNAL_FILE, sizeof(JOURNAL_KEY_ND_JOURNAL_FILE) - 1, njf->filename, njf->filename_len);
237
-
238
- NSD_JOURNAL_FOREACH_DATA(j, data, length)
239
- {
240
- const char *key, *value;
241
- size_t key_length, value_length;
242
-
243
- if (!parse_journal_field(data, length, &key, &key_length, &value, &value_length))
244
- continue;
245
-
246
-#ifdef NETDATA_INTERNAL_CHECKS
247
- usec_t origin_journal_ut = *msg_ut;
248
-#endif
249
- if (unlikely(
250
- key_length == sizeof(JD_SOURCE_REALTIME_TIMESTAMP) - 1 &&
251
- memcmp(key, JD_SOURCE_REALTIME_TIMESTAMP, sizeof(JD_SOURCE_REALTIME_TIMESTAMP) - 1) == 0)) {
252
- usec_t ut = str2ull(value, NULL);
253
- if (ut && ut < *msg_ut) {
254
- usec_t delta = *msg_ut - ut;
255
- *msg_ut = ut;
256
-
257
- if (delta > JOURNAL_VS_REALTIME_DELTA_MAX_UT)
258
- delta = JOURNAL_VS_REALTIME_DELTA_MAX_UT;
259
-
260
- // update max_journal_vs_realtime_delta_ut if the delta increased
261
- usec_t expected = njf->max_journal_vs_realtime_delta_ut;
262
- do {
263
- if (delta <= expected)
264
- break;
265
- } while (!__atomic_compare_exchange_n(
266
- &njf->max_journal_vs_realtime_delta_ut, &expected, delta, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
267
-
268
- internal_error(
269
- delta > expected,
270
- "increased max_journal_vs_realtime_delta_ut from %" PRIu64 " to %" PRIu64 ", "
271
- "journal %" PRIu64 ", actual %" PRIu64 " (delta %" PRIu64 ")",
272
- expected,
273
- delta,
274
- origin_journal_ut,
275
- *msg_ut,
276
- origin_journal_ut - (*msg_ut));
277
- }
278
- }
279
-
280
- bytes += length;
281
- facets_add_key_value_length(
282
- facets,
283
- key,
284
- key_length,
285
- value,
286
- value_length <= FACET_MAX_VALUE_LENGTH ? value_length : FACET_MAX_VALUE_LENGTH);
287
- }
288
-
289
- return bytes;
290
-}
291
-
292
-#define FUNCTION_PROGRESS_UPDATE_ROWS(rows_read, rows) __atomic_fetch_add(&(rows_read), rows, __ATOMIC_RELAXED)
293
-#define FUNCTION_PROGRESS_UPDATE_BYTES(bytes_read, bytes) __atomic_fetch_add(&(bytes_read), bytes, __ATOMIC_RELAXED)
294
-#define FUNCTION_PROGRESS_EVERY_ROWS (1ULL << 13)
295
-#define FUNCTION_DATA_ONLY_CHECK_EVERY_ROWS (1ULL << 7)
296
-
297
-static inline ND_SD_JOURNAL_STATUS check_stop(const bool *cancelled, const usec_t *stop_monotonic_ut)
298
-{
299
- if (cancelled && __atomic_load_n(cancelled, __ATOMIC_RELAXED)) {
300
- internal_error(true, "Function has been cancelled");
301
- return ND_SD_JOURNAL_CANCELLED;
302
- }
303
-
304
- if (now_monotonic_usec() > __atomic_load_n(stop_monotonic_ut, __ATOMIC_RELAXED)) {
305
- internal_error(true, "Function timed out");
306
- return ND_SD_JOURNAL_TIMED_OUT;
307
- }
308
-
309
- return ND_SD_JOURNAL_OK;
310
-}
311
-
312
-ND_SD_JOURNAL_STATUS nd_sd_journal_query_backward(
313
- NsdJournal *j,
314
- BUFFER *wb __maybe_unused,
315
- FACETS *facets,
316
- struct nd_journal_file *njf,
317
- LOGS_QUERY_STATUS *fqs)
318
-{
319
- usec_t anchor_delta = __atomic_load_n(&njf->max_journal_vs_realtime_delta_ut, __ATOMIC_RELAXED);
320
- lqs_query_timeframe(fqs, anchor_delta);
321
- usec_t start_ut = fqs->query.start_ut;
322
- usec_t stop_ut = fqs->query.stop_ut;
323
- bool stop_when_full = fqs->query.stop_when_full;
324
-
325
- fqs->c.query_file.start_ut = start_ut;
326
- fqs->c.query_file.stop_ut = stop_ut;
327
-
328
- if (!nd_sd_journal_seek_to(j, start_ut))
329
- return ND_SD_JOURNAL_FAILED_TO_SEEK;
330
-
331
- size_t errors_no_timestamp = 0;
332
- usec_t latest_msg_ut = 0; // the biggest timestamp we have seen so far
333
- usec_t first_msg_ut = 0; // the first message we got from the db
334
- size_t row_counter = 0, last_row_counter = 0, rows_useful = 0;
335
- size_t bytes = 0, last_bytes = 0;
336
-
337
- usec_t last_usec_from = 0;
338
- usec_t last_usec_to = 0;
339
-
340
- ND_SD_JOURNAL_STATUS status = ND_SD_JOURNAL_OK;
341
-
342
- facets_rows_begin(facets);
343
- while (status == ND_SD_JOURNAL_OK && nsd_journal_previous(j) > 0) {
344
- usec_t msg_ut = 0;
345
- if (nsd_journal_get_realtime_usec(j, &msg_ut) < 0 || !msg_ut) {
346
- errors_no_timestamp++;
347
- continue;
348
- }
349
-
350
- if (unlikely(msg_ut > start_ut))
351
- continue;
352
-
353
- if (unlikely(msg_ut < stop_ut))
354
- break;
355
-
356
- if (unlikely(msg_ut > latest_msg_ut))
357
- latest_msg_ut = msg_ut;
358
-
359
- if (unlikely(!first_msg_ut)) {
360
- first_msg_ut = msg_ut;
361
- fqs->c.query_file.first_msg_ut = msg_ut;
362
-
363
-#ifdef HAVE_SD_JOURNAL_GET_SEQNUM
364
- if (nsd_journal_get_seqnum(j, &fqs->c.query_file.first_msg_seqnum, &fqs->c.query_file.first_msg_writer) <
365
- 0) {
366
- fqs->c.query_file.first_msg_seqnum = 0;
367
- fqs->c.query_file.first_msg_writer = NSD_ID128_NULL;
368
- }
369
-#endif
370
- }
371
-
372
- sampling_t sample = is_row_in_sample(
373
- j, fqs, njf, msg_ut, FACETS_ANCHOR_DIRECTION_BACKWARD, facets_row_candidate_to_keep(facets, msg_ut));
374
-
375
- if (sample == SAMPLING_FULL) {
376
- bytes += nd_sd_journal_process_row(j, facets, njf, &msg_ut);
377
-
378
- // make sure each line gets a unique timestamp
379
- if (unlikely(msg_ut >= last_usec_from && msg_ut <= last_usec_to))
380
- msg_ut = --last_usec_from;
381
- else
382
- last_usec_from = last_usec_to = msg_ut;
383
-
384
- if (facets_row_finished(facets, msg_ut))
385
- rows_useful++;
386
-
387
- row_counter++;
388
- if (unlikely(
389
- (row_counter % FUNCTION_DATA_ONLY_CHECK_EVERY_ROWS) == 0 && stop_when_full &&
390
- facets_rows(facets) >= fqs->rq.entries)) {
391
- // stop the data only query
392
- usec_t oldest = facets_row_oldest_ut(facets);
393
- if (oldest && msg_ut < (oldest - anchor_delta))
394
- break;
395
- }
396
-
397
- if (unlikely(row_counter % FUNCTION_PROGRESS_EVERY_ROWS == 0)) {
398
- FUNCTION_PROGRESS_UPDATE_ROWS(fqs->c.rows_read, row_counter - last_row_counter);
399
- last_row_counter = row_counter;
400
-
401
- FUNCTION_PROGRESS_UPDATE_BYTES(fqs->c.bytes_read, bytes - last_bytes);
402
- last_bytes = bytes;
403
-
404
- status = check_stop(fqs->cancelled, fqs->stop_monotonic_ut);
405
- }
406
- } else if (sample == SAMPLING_SKIP_FIELDS)
407
- facets_row_finished_unsampled(facets, msg_ut);
408
- else {
409
- sampling_update_running_query_file_estimates(facets, j, fqs, njf, msg_ut, FACETS_ANCHOR_DIRECTION_BACKWARD);
410
- break;
411
- }
412
- }
413
-
414
- FUNCTION_PROGRESS_UPDATE_ROWS(fqs->c.rows_read, row_counter - last_row_counter);
415
- FUNCTION_PROGRESS_UPDATE_BYTES(fqs->c.bytes_read, bytes - last_bytes);
416
-
417
- fqs->c.rows_useful += rows_useful;
418
-
419
- if (errors_no_timestamp)
420
- netdata_log_error("SYSTEMD-JOURNAL: %zu lines did not have timestamps", errors_no_timestamp);
421
-
422
- if (latest_msg_ut > fqs->last_modified)
423
- fqs->last_modified = latest_msg_ut;
424
-
425
- return status;
426
-}
427
-
428
-ND_SD_JOURNAL_STATUS nd_sd_journal_query_forward(
429
- NsdJournal *j,
430
- BUFFER *wb __maybe_unused,
431
- FACETS *facets,
432
- struct nd_journal_file *njf,
433
- LOGS_QUERY_STATUS *fqs)
434
-{
435
- usec_t anchor_delta = __atomic_load_n(&njf->max_journal_vs_realtime_delta_ut, __ATOMIC_RELAXED);
436
- lqs_query_timeframe(fqs, anchor_delta);
437
- usec_t start_ut = fqs->query.start_ut;
438
- usec_t stop_ut = fqs->query.stop_ut;
439
- bool stop_when_full = fqs->query.stop_when_full;
440
-
441
- fqs->c.query_file.start_ut = start_ut;
442
- fqs->c.query_file.stop_ut = stop_ut;
443
-
444
- if (!nd_sd_journal_seek_to(j, start_ut))
445
- return ND_SD_JOURNAL_FAILED_TO_SEEK;
446
-
447
- size_t errors_no_timestamp = 0;
448
- usec_t latest_msg_ut = 0; // the biggest timestamp we have seen so far
449
- usec_t first_msg_ut = 0; // the first message we got from the db
450
- size_t row_counter = 0, last_row_counter = 0, rows_useful = 0;
451
- size_t bytes = 0, last_bytes = 0;
452
-
453
- usec_t last_usec_from = 0;
454
- usec_t last_usec_to = 0;
455
-
456
- ND_SD_JOURNAL_STATUS status = ND_SD_JOURNAL_OK;
457
-
458
- facets_rows_begin(facets);
459
- while (status == ND_SD_JOURNAL_OK && nsd_journal_next(j) > 0) {
460
- usec_t msg_ut = 0;
461
- if (nsd_journal_get_realtime_usec(j, &msg_ut) < 0 || !msg_ut) {
462
- errors_no_timestamp++;
463
- continue;
464
- }
465
-
466
- if (unlikely(msg_ut < start_ut))
467
- continue;
468
-
469
- if (unlikely(msg_ut > stop_ut))
470
- break;
471
-
472
- if (likely(msg_ut > latest_msg_ut))
473
- latest_msg_ut = msg_ut;
474
-
475
- if (unlikely(!first_msg_ut)) {
476
- first_msg_ut = msg_ut;
477
- fqs->c.query_file.first_msg_ut = msg_ut;
478
- }
479
-
480
- sampling_t sample = is_row_in_sample(
481
- j, fqs, njf, msg_ut, FACETS_ANCHOR_DIRECTION_FORWARD, facets_row_candidate_to_keep(facets, msg_ut));
482
-
483
- if (sample == SAMPLING_FULL) {
484
- bytes += nd_sd_journal_process_row(j, facets, njf, &msg_ut);
485
-
486
- // make sure each line gets a unique timestamp
487
- if (unlikely(msg_ut >= last_usec_from && msg_ut <= last_usec_to))
488
- msg_ut = ++last_usec_to;
489
- else
490
- last_usec_from = last_usec_to = msg_ut;
491
-
492
- if (facets_row_finished(facets, msg_ut))
493
- rows_useful++;
494
-
495
- row_counter++;
496
- if (unlikely(
497
- (row_counter % FUNCTION_DATA_ONLY_CHECK_EVERY_ROWS) == 0 && stop_when_full &&
498
- facets_rows(facets) >= fqs->rq.entries)) {
499
- // stop the data only query
500
- usec_t newest = facets_row_newest_ut(facets);
501
- if (newest && msg_ut > (newest + anchor_delta))
502
- break;
503
- }
504
-
505
- if (unlikely(row_counter % FUNCTION_PROGRESS_EVERY_ROWS == 0)) {
506
- FUNCTION_PROGRESS_UPDATE_ROWS(fqs->c.rows_read, row_counter - last_row_counter);
507
- last_row_counter = row_counter;
508
-
509
- FUNCTION_PROGRESS_UPDATE_BYTES(fqs->c.bytes_read, bytes - last_bytes);
510
- last_bytes = bytes;
511
-
512
- status = check_stop(fqs->cancelled, fqs->stop_monotonic_ut);
513
- }
514
- } else if (sample == SAMPLING_SKIP_FIELDS)
515
- facets_row_finished_unsampled(facets, msg_ut);
516
- else {
517
- sampling_update_running_query_file_estimates(facets, j, fqs, njf, msg_ut, FACETS_ANCHOR_DIRECTION_FORWARD);
518
- break;
519
- }
520
- }
521
-
522
- FUNCTION_PROGRESS_UPDATE_ROWS(fqs->c.rows_read, row_counter - last_row_counter);
523
- FUNCTION_PROGRESS_UPDATE_BYTES(fqs->c.bytes_read, bytes - last_bytes);
524
-
525
- fqs->c.rows_useful += rows_useful;
526
-
527
- if (errors_no_timestamp)
528
- netdata_log_error("SYSTEMD-JOURNAL: %zu lines did not have timestamps", errors_no_timestamp);
529
-
530
- if (latest_msg_ut > fqs->last_modified)
531
- fqs->last_modified = latest_msg_ut;
532
-
533
- return status;
534
-}
535
-
536
-bool nd_sd_journal_check_if_modified_since(NsdJournal *j, usec_t seek_to, usec_t last_modified)
537
-{
538
- // return true, if data have been modified since the timestamp
539
-
540
- if (!last_modified || !seek_to)
541
- return false;
542
-
543
- if (!nd_sd_journal_seek_to(j, seek_to))
544
- return false;
545
-
546
- usec_t first_msg_ut = 0;
547
- while (nsd_journal_previous(j) > 0) {
548
- usec_t msg_ut;
549
- if (nsd_journal_get_realtime_usec(j, &msg_ut) < 0)
550
- continue;
551
-
552
- first_msg_ut = msg_ut;
553
- break;
554
- }
555
-
556
- return first_msg_ut != last_modified;
557
-}
558
-
559
-#ifdef HAVE_SD_JOURNAL_RESTART_FIELDS
560
-static bool netdata_systemd_filtering_by_journal(NsdJournal *j, FACETS *facets, LOGS_QUERY_STATUS *lqs)
561
-{
562
- const char *field = NULL;
563
- const void *data = NULL;
564
- size_t data_length;
565
- size_t added_keys = 0;
566
- size_t failures = 0;
567
- size_t filters_added = 0;
568
-
569
- NSD_JOURNAL_FOREACH_FIELD(j, field)
570
- { // for each key
571
- bool interesting;
572
-
573
- if (lqs->rq.data_only)
574
- interesting = facets_key_name_is_filter(facets, field);
575
- else
576
- interesting = facets_key_name_is_facet(facets, field);
577
-
578
- if (interesting) {
579
- int r = nsd_journal_query_unique(j, field);
580
- if (r >= 0) {
581
- bool added_this_key = false;
582
- size_t added_values = 0;
583
-
584
- nsd_journal_restart_unique(j);
585
- while ((r = nsd_journal_enumerate_available_unique(j, &data, &data_length)) > 0) {
586
- // for each value of the key
587
- const char *key, *value;
588
- size_t key_length, value_length;
589
-
590
- if (!parse_journal_field(data, data_length, &key, &key_length, &value, &value_length))
591
- continue;
592
-
593
- facets_add_possible_value_name_to_key(facets, key, key_length, value, value_length);
594
-
595
- if (!facets_key_name_value_length_is_selected(facets, key, key_length, value, value_length))
596
- continue;
597
-
598
- if (added_keys && !added_this_key) {
599
- if (nsd_journal_add_conjunction(j) < 0) // key AND key AND key
600
- failures++;
601
-
602
- added_this_key = true;
603
- added_keys++;
604
- } else if (added_values)
605
- if (nsd_journal_add_disjunction(j) < 0) // value OR value OR value
606
- failures++;
607
-
608
- if (nsd_journal_add_match(j, data, data_length) < 0)
609
- failures++;
610
-
611
- if (!added_keys) {
612
- added_keys++;
613
- added_this_key = true;
614
- }
615
-
616
- added_values++;
617
- filters_added++;
618
- }
619
-
620
- if (r < 0)
621
- failures++;
622
- } else {
623
- failures++;
624
- }
625
- }
626
- }
627
-
628
- if (failures) {
629
- lqs_log_error(lqs, "failed to setup journal filter, will run the full query.");
630
- nsd_journal_flush_matches(j);
631
- return true;
632
- }
633
-
634
- return filters_added ? true : false;
635
-}
636
-#endif // HAVE_SD_JOURNAL_RESTART_FIELDS
637
-
638
-static ND_SD_JOURNAL_STATUS nd_sd_journal_query_one_file(
639
- const char *filename,
640
- BUFFER *wb,
641
- FACETS *facets,
642
- struct nd_journal_file *njf,
643
- LOGS_QUERY_STATUS *fqs)
644
-{
645
- NsdJournal *j = NULL;
646
- errno_clear();
647
-
648
- fstat_cache_enable_on_thread();
649
-
650
- const char *paths[2] = {
651
- [0] = filename,
652
- [1] = NULL,
653
- };
654
-
655
- if (nsd_journal_open_files(&j, paths, ND_SD_JOURNAL_OPEN_FLAGS) < 0 || !j) {
656
- netdata_log_error("JOURNAL: cannot open file '%s' for query", filename);
657
- fstat_cache_disable_on_thread();
658
- return ND_SD_JOURNAL_FAILED_TO_OPEN;
659
- }
660
-
661
- ND_SD_JOURNAL_STATUS status;
662
- bool matches_filters = true;
663
-
664
-#ifdef HAVE_SD_JOURNAL_RESTART_FIELDS
665
- if (fqs->rq.slice) {
666
- usec_t started = now_monotonic_usec();
667
-
668
- matches_filters = netdata_systemd_filtering_by_journal(j, facets, fqs) || !fqs->rq.filters;
669
- usec_t ended = now_monotonic_usec();
670
-
671
- fqs->c.matches_setup_ut += (ended - started);
672
- }
673
-#endif // HAVE_SD_JOURNAL_RESTART_FIELDS
674
-
675
- if (matches_filters) {
676
- if (fqs->rq.direction == FACETS_ANCHOR_DIRECTION_FORWARD)
677
- status = nd_sd_journal_query_forward(j, wb, facets, njf, fqs);
678
- else
679
- status = nd_sd_journal_query_backward(j, wb, facets, njf, fqs);
680
- } else
681
- status = ND_SD_JOURNAL_NO_FILE_MATCHED;
682
-
683
- nsd_journal_close(j);
684
- fstat_cache_disable_on_thread();
156
+#include "systemd-journal-execute.h"
157
686
- return status;
687
-}
688
-
689
-static bool jf_is_mine(struct nd_journal_file *njf, LOGS_QUERY_STATUS *fqs)
690
-{
691
- if ((fqs->rq.source_type == ND_SD_JF_NONE && !fqs->rq.sources) || (njf->source_type & fqs->rq.source_type) ||
692
- (fqs->rq.sources && simple_pattern_matches(fqs->rq.sources, string2str(njf->source)))) {
693
- if (!njf->msg_last_ut)
694
- // the file is not scanned yet, or the timestamps have not been updated,
695
- // so we don't know if it can contribute or not - let's add it.
696
- return true;
697
-
698
- usec_t anchor_delta = JOURNAL_VS_REALTIME_DELTA_MAX_UT;
699
- usec_t first_ut = njf->msg_first_ut - anchor_delta;
700
- usec_t last_ut = njf->msg_last_ut + anchor_delta;
701
-
702
- if (last_ut >= fqs->rq.after_ut && first_ut <= fqs->rq.before_ut)
703
- return true;
704
- }
705
-
706
- return false;
707
-}
708
-
709
-static int nd_sd_journal_query(BUFFER *wb, LOGS_QUERY_STATUS *lqs)
710
-{
711
- FACETS *facets = lqs->facets;
712
-
713
- ND_SD_JOURNAL_STATUS status = ND_SD_JOURNAL_NO_FILE_MATCHED;
714
- struct nd_journal_file *njf;
715
-
716
- lqs->c.files_matched = 0;
717
- lqs->c.file_working = 0;
718
- lqs->c.rows_useful = 0;
719
- lqs->c.rows_read = 0;
720
- lqs->c.bytes_read = 0;
721
-
722
- size_t files_used = 0;
723
- size_t files_max = dictionary_entries(nd_journal_files_registry);
724
- const DICTIONARY_ITEM *file_items[files_max];
725
-
726
- // count the files
727
- bool files_are_newer = false;
728
- dfe_start_read(nd_journal_files_registry, njf)
729
- {
730
- if (!jf_is_mine(njf, lqs))
731
- continue;
732
-
733
- file_items[files_used++] = dictionary_acquired_item_dup(nd_journal_files_registry, njf_dfe.item);
734
-
735
- if (njf->msg_last_ut > lqs->rq.if_modified_since)
736
- files_are_newer = true;
737
- }
738
- dfe_done(jf);
739
-
740
- lqs->c.files_matched = files_used;
741
-
742
- if (lqs->rq.if_modified_since && !files_are_newer) {
743
- // release the files
744
- for (size_t f = 0; f < files_used; f++)
745
- dictionary_acquired_item_release(nd_journal_files_registry, file_items[f]);
746
-
747
- return rrd_call_function_error(wb, "No new data since the previous call.", HTTP_RESP_NOT_MODIFIED);
748
- }
749
-
750
- // sort the files, so that they are optimal for facets
751
- if (files_used >= 2) {
752
- if (lqs->rq.direction == FACETS_ANCHOR_DIRECTION_BACKWARD)
753
- qsort(file_items, files_used, sizeof(const DICTIONARY_ITEM *), nd_journal_file_dict_items_backward_compar);
754
- else
755
- qsort(file_items, files_used, sizeof(const DICTIONARY_ITEM *), nd_journal_file_dict_items_forward_compar);
756
- }
757
-
758
- bool partial = false;
759
- usec_t query_started_ut = now_monotonic_usec();
760
- usec_t started_ut = query_started_ut;
761
- usec_t ended_ut = started_ut;
762
- usec_t duration_ut = 0, max_duration_ut = 0;
763
- usec_t progress_duration_ut = 0;
764
-
765
- sampling_query_init(lqs, facets);
766
-
767
- buffer_json_member_add_array(wb, "_journal_files");
768
- for (size_t f = 0; f < files_used; f++) {
769
- const char *filename = dictionary_acquired_item_name(file_items[f]);
770
- njf = dictionary_acquired_item_value(file_items[f]);
771
-
772
- if (!jf_is_mine(njf, lqs))
773
- continue;
774
-
775
- started_ut = ended_ut;
776
-
777
- // do not even try to do the query if we expect it to pass the timeout
778
- if (ended_ut + max_duration_ut * 3 >= *lqs->stop_monotonic_ut) {
779
- partial = true;
780
- status = ND_SD_JOURNAL_TIMED_OUT;
781
- break;
782
- }
783
-
784
- lqs->c.file_working++;
785
-
786
- size_t fs_calls = fstat_thread_calls;
787
- size_t fs_cached = fstat_thread_cached_responses;
788
- size_t rows_useful = lqs->c.rows_useful;
789
- size_t rows_read = lqs->c.rows_read;
790
- size_t bytes_read = lqs->c.bytes_read;
791
- size_t matches_setup_ut = lqs->c.matches_setup_ut;
792
-
793
- sampling_file_init(lqs, njf);
794
-
795
- ND_SD_JOURNAL_STATUS tmp_status = nd_sd_journal_query_one_file(filename, wb, facets, njf, lqs);
796
-
797
- rows_useful = lqs->c.rows_useful - rows_useful;
798
- rows_read = lqs->c.rows_read - rows_read;
799
- bytes_read = lqs->c.bytes_read - bytes_read;
800
- matches_setup_ut = lqs->c.matches_setup_ut - matches_setup_ut;
801
- fs_calls = fstat_thread_calls - fs_calls;
802
- fs_cached = fstat_thread_cached_responses - fs_cached;
803
-
804
- ended_ut = now_monotonic_usec();
805
- duration_ut = ended_ut - started_ut;
806
-
807
- if (duration_ut > max_duration_ut)
808
- max_duration_ut = duration_ut;
809
-
810
- progress_duration_ut += duration_ut;
811
- if (progress_duration_ut >= ND_SD_JOURNAL_PROGRESS_EVERY_UT) {
812
- progress_duration_ut = 0;
813
- netdata_mutex_lock(&stdout_mutex);
814
- pluginsd_function_progress_to_stdout(lqs->rq.transaction, f + 1, files_used);
815
- netdata_mutex_unlock(&stdout_mutex);
816
- }
817
-
818
- buffer_json_add_array_item_object(wb); // journal file
819
- {
820
- // information about the file
821
- buffer_json_member_add_string(wb, "_filename", filename);
822
- buffer_json_member_add_uint64(wb, "_source_type", njf->source_type);
823
- buffer_json_member_add_string(wb, "_source", string2str(njf->source));
824
- buffer_json_member_add_uint64(wb, "_last_modified_ut", njf->file_last_modified_ut);
825
- buffer_json_member_add_uint64(wb, "_msg_first_ut", njf->msg_first_ut);
826
- buffer_json_member_add_uint64(wb, "_msg_last_ut", njf->msg_last_ut);
827
- buffer_json_member_add_uint64(wb, "_journal_vs_realtime_delta_ut", njf->max_journal_vs_realtime_delta_ut);
828
-
829
- // information about the current use of the file
830
- buffer_json_member_add_uint64(wb, "duration_ut", ended_ut - started_ut);
831
- buffer_json_member_add_uint64(wb, "rows_read", rows_read);
832
- buffer_json_member_add_uint64(wb, "rows_useful", rows_useful);
833
- buffer_json_member_add_double(
834
- wb, "rows_per_second", (double)rows_read / (double)duration_ut * (double)USEC_PER_SEC);
835
- buffer_json_member_add_uint64(wb, "bytes_read", bytes_read);
836
- buffer_json_member_add_double(
837
- wb, "bytes_per_second", (double)bytes_read / (double)duration_ut * (double)USEC_PER_SEC);
838
- buffer_json_member_add_uint64(wb, "duration_matches_ut", matches_setup_ut);
839
- buffer_json_member_add_uint64(wb, "fstat_query_calls", fs_calls);
840
- buffer_json_member_add_uint64(wb, "fstat_query_cached_responses", fs_cached);
841
-
842
- if (lqs->rq.sampling) {
843
- buffer_json_member_add_object(wb, "_sampling");
844
- {
845
- buffer_json_member_add_uint64(wb, "sampled", lqs->c.samples_per_file.sampled);
846
- buffer_json_member_add_uint64(wb, "unsampled", lqs->c.samples_per_file.unsampled);
847
- buffer_json_member_add_uint64(wb, "estimated", lqs->c.samples_per_file.estimated);
848
- }
849
- buffer_json_object_close(wb); // _sampling
850
- }
851
- }
852
- buffer_json_object_close(wb); // journal file
853
-
854
- bool stop = false;
855
- switch (tmp_status) {
856
- case ND_SD_JOURNAL_OK:
857
- case ND_SD_JOURNAL_NO_FILE_MATCHED:
858
- status = (status == ND_SD_JOURNAL_OK) ? ND_SD_JOURNAL_OK : tmp_status;
859
- break;
860
-
861
- case ND_SD_JOURNAL_FAILED_TO_OPEN:
862
- case ND_SD_JOURNAL_FAILED_TO_SEEK:
863
- partial = true;
864
- if (status == ND_SD_JOURNAL_NO_FILE_MATCHED)
865
- status = tmp_status;
866
- break;
867
-
868
- case ND_SD_JOURNAL_CANCELLED:
869
- case ND_SD_JOURNAL_TIMED_OUT:
870
- partial = true;
871
- stop = true;
872
- status = tmp_status;
873
- break;
874
-
875
- case ND_SD_JOURNAL_NOT_MODIFIED:
876
- internal_fatal(true, "this should never be returned here");
877
- break;
878
- }
879
-
880
- if (stop)
881
- break;
882
- }
883
- buffer_json_array_close(wb); // _journal_files
884
-
885
- // release the files
886
- for (size_t f = 0; f < files_used; f++)
887
- dictionary_acquired_item_release(nd_journal_files_registry, file_items[f]);
888
-
889
- switch (status) {
890
- case ND_SD_JOURNAL_OK:
891
- if (lqs->rq.if_modified_since && !lqs->c.rows_useful)
892
- return rrd_call_function_error(
893
- wb, "No additional useful data since the previous call.", HTTP_RESP_NOT_MODIFIED);
894
- break;
895
-
896
- case ND_SD_JOURNAL_TIMED_OUT:
897
- case ND_SD_JOURNAL_NO_FILE_MATCHED:
898
- break;
899
-
900
- case ND_SD_JOURNAL_CANCELLED:
901
- return rrd_call_function_error(wb, "Request cancelled.", HTTP_RESP_CLIENT_CLOSED_REQUEST);
902
-
903
- case ND_SD_JOURNAL_NOT_MODIFIED:
904
- return rrd_call_function_error(wb, "No new data since the previous call.", HTTP_RESP_NOT_MODIFIED);
905
-
906
- case ND_SD_JOURNAL_FAILED_TO_OPEN:
907
- return rrd_call_function_error(wb, "Failed to open systemd journal file.", HTTP_RESP_INTERNAL_SERVER_ERROR);
908
-
909
- case ND_SD_JOURNAL_FAILED_TO_SEEK:
910
- return rrd_call_function_error(
911
- wb, "Failed to seek in systemd journal file.", HTTP_RESP_INTERNAL_SERVER_ERROR);
912
-
913
- default:
914
- return rrd_call_function_error(wb, "Unknown status", HTTP_RESP_INTERNAL_SERVER_ERROR);
915
- }
916
-
917
- buffer_json_member_add_uint64(wb, "status", HTTP_RESP_OK);
918
- buffer_json_member_add_boolean(wb, "partial", partial);
919
- buffer_json_member_add_string(wb, "type", "table");
920
-
921
- // build a message for the query
922
- if (!lqs->rq.data_only) {
923
- CLEAN_BUFFER *msg = buffer_create(0, NULL);
924
- CLEAN_BUFFER *msg_description = buffer_create(0, NULL);
925
- ND_LOG_FIELD_PRIORITY msg_priority = NDLP_INFO;
926
-
927
- if (!nd_journal_files_completed_once()) {
928
- buffer_strcat(msg, "Journals are still being scanned. ");
929
- buffer_strcat(
930
- msg_description,
931
- "LIBRARY SCAN: The journal files are still being scanned, you are probably viewing incomplete data. ");
932
- msg_priority = NDLP_WARNING;
933
- }
934
-
935
- if (partial) {
936
- buffer_strcat(msg, "Query timed-out, incomplete data. ");
937
- buffer_strcat(
938
- msg_description,
939
- "QUERY TIMEOUT: The query timed out and may not include all the data of the selected window. ");
940
- msg_priority = NDLP_WARNING;
941
- }
942
-
943
- if (lqs->c.samples.estimated || lqs->c.samples.unsampled) {
944
- double percent = (double)(lqs->c.samples.sampled * 100.0 /
945
- (lqs->c.samples.estimated + lqs->c.samples.unsampled + lqs->c.samples.sampled));
946
- buffer_sprintf(msg, "%.2f%% real data", percent);
947
- buffer_sprintf(msg_description, "ACTUAL DATA: The filters counters reflect %0.2f%% of the data. ", percent);
948
- msg_priority = MIN(msg_priority, NDLP_NOTICE);
949
- }
950
-
951
- if (lqs->c.samples.unsampled) {
952
- double percent = (double)(lqs->c.samples.unsampled * 100.0 /
953
- (lqs->c.samples.estimated + lqs->c.samples.unsampled + lqs->c.samples.sampled));
954
- buffer_sprintf(msg, ", %.2f%% unsampled", percent);
955
- buffer_sprintf(
956
- msg_description,
957
- "UNSAMPLED DATA: %0.2f%% of the events exist and have been counted, but their values have not been evaluated, so they are not included in the filters counters. ",
958
- percent);
959
- msg_priority = MIN(msg_priority, NDLP_NOTICE);
960
- }
961
-
962
- if (lqs->c.samples.estimated) {
963
- double percent = (double)(lqs->c.samples.estimated * 100.0 /
964
- (lqs->c.samples.estimated + lqs->c.samples.unsampled + lqs->c.samples.sampled));
965
- buffer_sprintf(msg, ", %.2f%% estimated", percent);
966
- buffer_sprintf(
967
- msg_description,
968
- "ESTIMATED DATA: The query selected a large amount of data, so to avoid delaying too much, the presented data are estimated by %0.2f%%. ",
969
- percent);
970
- msg_priority = MIN(msg_priority, NDLP_NOTICE);
971
- }
972
-
973
- buffer_json_member_add_object(wb, "message");
974
- if (buffer_tostring(msg)) {
975
- buffer_json_member_add_string(wb, "title", buffer_tostring(msg));
976
- buffer_json_member_add_string(wb, "description", buffer_tostring(msg_description));
977
- buffer_json_member_add_string(wb, "status", nd_log_id2priority(msg_priority));
978
- }
979
- // else send an empty object if there is nothing to tell
980
- buffer_json_object_close(wb); // message
981
- }
982
-
983
- if (!lqs->rq.data_only) {
984
- buffer_json_member_add_time_t(wb, "update_every", 1);
985
- buffer_json_member_add_string(wb, "help", ND_SD_JOURNAL_FUNCTION_DESCRIPTION);
986
- }
987
-
988
- if (!lqs->rq.data_only || lqs->rq.tail)
989
- buffer_json_member_add_uint64(wb, "last_modified", lqs->last_modified);
990
-
991
- facets_sort_and_reorder_keys(facets);
992
- facets_report(facets, wb, used_hashes_registry);
993
-
994
- wb->expires = now_realtime_sec() + (lqs->rq.data_only ? 3600 : 0);
995
- buffer_json_member_add_time_t(wb, "expires", wb->expires);
996
-
997
- buffer_json_member_add_object(wb, "_fstat_caching");
998
- {
999
- buffer_json_member_add_uint64(wb, "calls", fstat_thread_calls);
1000
- buffer_json_member_add_uint64(wb, "cached", fstat_thread_cached_responses);
1001
- }
1002
- buffer_json_object_close(wb); // _fstat_caching
1003
-
1004
- if (lqs->rq.sampling) {
1005
- buffer_json_member_add_object(wb, "_sampling");
1006
- {
1007
- buffer_json_member_add_uint64(wb, "sampled", lqs->c.samples.sampled);
1008
- buffer_json_member_add_uint64(wb, "unsampled", lqs->c.samples.unsampled);
1009
- buffer_json_member_add_uint64(wb, "estimated", lqs->c.samples.estimated);
1010
- }
1011
- buffer_json_object_close(wb); // _sampling
1012
- }
1013
-
1014
- wb->content_type = CT_APPLICATION_JSON;
1015
- wb->response_code = HTTP_RESP_OK;
1016
- return wb->response_code;
1017
-}
1018
-
1019
-static void systemd_journal_register_transformations(LOGS_QUERY_STATUS *lqs)
1020
-{
158
+static
159
+void systemd_journal_register_transformations(LOGS_QUERY_STATUS *lqs) {
160
FACETS *facets = lqs->facets;
161
LOGS_QUERY_REQUEST *rq = &lqs->rq;
162
src/libnetdata/facets/logs_query_status.h
+1
-1
@@ -544,7 +544,7 @@ static inline bool lqs_request_parse_GET(LOGS_QUERY_STATUS *lqs, BUFFER *wb, cha
544
rq->slice = true;
545
}
546
else if(strncmp(keyword, LQS_PARAMETER_SOURCE ":", sizeof(LQS_PARAMETER_SOURCE ":") - 1) == 0) {
547
- const char *value = &keyword[sizeof(LQS_PARAMETER_SOURCE ":") - 1];
547
+ char *value = &keyword[sizeof(LQS_PARAMETER_SOURCE ":") - 1];
548
549
buffer_json_member_add_array(wb, LQS_PARAMETER_SOURCE);
550