9
#include "libnetdata/libnetdata.h"
10
#include "libnetdata/required_dummies.h"
11
12
+#include <linux/capability.h>
13
#include <systemd/sd-journal.h>
14
#include <syslog.h>
15
16
+// ----------------------------------------------------------------------------
17
+// fstat64 overloading to speed up libsystemd
18
+// https://github.com/systemd/systemd/pull/29261
19
+
20
+#define ND_SD_JOURNAL_OPEN_FLAGS (0)
21
+
22
+#ifdef HAVE_SD_JOURNAL_OPEN_FILES_FD
23
+
24
+#include <dlfcn.h>
25
+#include <sys/stat.h>
26
+
27
+#define FSTAT_CACHE_MAX 1024
28
+struct fdstat64_cache_entry {
29
+ bool enabled;
30
+ bool updated;
31
+ int err_no;
32
+ struct stat64 stat;
33
+ int ret;
34
+ size_t cached_count;
35
+};
36
+struct fdstat64_cache_entry fstat64_cache[FSTAT_CACHE_MAX] = {0 };
37
+
38
+static void fstat_cache_enable(int fd) {
39
+ if(fd >= 0 && fd < FSTAT_CACHE_MAX) {
40
+ fstat64_cache[fd].enabled = true;
41
+ fstat64_cache[fd].updated = false;
42
+ fstat64_cache[fd].cached_count = 0;
43
+ }
44
+}
45
+
46
+static size_t fstat_cache_disable(int fd) {
47
+ size_t cached_count = 0;
48
+
49
+ if(fd >= 0 && fd < FSTAT_CACHE_MAX) {
50
+ fstat64_cache[fd].enabled = false;
51
+ fstat64_cache[fd].updated = false;
52
+ cached_count = fstat64_cache[fd].cached_count;
53
+ fstat64_cache[fd].cached_count = 0;
54
+ }
55
+
56
+ return cached_count;
57
+}
58
+
59
+static size_t fstat_calls = 0;
60
+static size_t fstat_cached_responses = 0;
61
+
62
+int fstat64(int fd, struct stat64 *buf) {
63
+ static int (*real_fstat)(int, struct stat64 *) = NULL;
64
+ if (!real_fstat)
65
+ real_fstat = dlsym(RTLD_NEXT, "fstat64");
66
+
67
+ fstat_calls++;
68
+
69
+ if(fd >= 0 && fd < FSTAT_CACHE_MAX && fstat64_cache[fd].enabled && fstat64_cache[fd].updated) {
70
+ fstat_cached_responses++;
71
+ errno = fstat64_cache[fd].err_no;
72
+ *buf = fstat64_cache[fd].stat;
73
+ fstat64_cache[fd].cached_count++;
74
+ return fstat64_cache[fd].ret;
75
+ }
76
+
77
+ int ret = real_fstat(fd, buf);
78
+
79
+ if(fd >= 0 && fd < FSTAT_CACHE_MAX && fstat64_cache[fd].enabled) {
80
+ fstat64_cache[fd].ret = ret;
81
+ fstat64_cache[fd].updated = true;
82
+ fstat64_cache[fd].err_no = errno;
83
+ fstat64_cache[fd].stat = *buf;
84
+ }
85
+
86
+ return ret;
87
+}
88
+
89
+#endif // HAVE_SD_JOURNAL_OPEN_FILES_FD
90
+
91
+// ----------------------------------------------------------------------------
92
+
93
#define FACET_MAX_VALUE_LENGTH 8192
94
+#define SYSTEMD_JOURNAL_MAX_SOURCE_LEN 64
95
96
#define SYSTEMD_JOURNAL_FUNCTION_DESCRIPTION "View, search and analyze systemd journal entries."
97
#define SYSTEMD_JOURNAL_FUNCTION_NAME "systemd-journal"
19
-#define SYSTEMD_JOURNAL_DEFAULT_TIMEOUT 30
98
+#define SYSTEMD_JOURNAL_DEFAULT_TIMEOUT 60
99
#define SYSTEMD_JOURNAL_MAX_PARAMS 100
100
#define SYSTEMD_JOURNAL_DEFAULT_QUERY_DURATION (3 * 3600)
101
#define SYSTEMD_JOURNAL_DEFAULT_ITEMS_PER_QUERY 200
23
-#define SYSTEMD_JOURNAL_EXCESS_ROWS_ALLOWED 50
24
-#define SYSTEMD_JOURNAL_WORKER_THREADS 2
102
+#define SYSTEMD_JOURNAL_WORKER_THREADS 5
103
104
#define JOURNAL_PARAMETER_HELP "help"
105
#define JOURNAL_PARAMETER_AFTER "after"
114
#define JOURNAL_PARAMETER_DATA_ONLY "data_only"
115
#define JOURNAL_PARAMETER_SOURCE "source"
116
#define JOURNAL_PARAMETER_INFO "info"
117
+#define JOURNAL_PARAMETER_ID "id"
118
+#define JOURNAL_PARAMETER_PROGRESS "progress"
119
+#define JOURNAL_PARAMETER_SLICE "slice"
120
+#define JOURNAL_PARAMETER_DELTA "delta"
121
+#define JOURNAL_PARAMETER_TAIL "tail"
122
+
123
+#define JOURNAL_DEFAULT_SLICE_MODE true
124
+#define JOURNAL_DEFAULT_DIRECTION FACETS_ANCHOR_DIRECTION_BACKWARD
125
126
#define SYSTEMD_ALWAYS_VISIBLE_KEYS NULL
41
-#define SYSTEMD_KEYS_EXCLUDED_FROM_FACETS NULL
127
+
128
+#define SYSTEMD_KEYS_EXCLUDED_FROM_FACETS \
129
+ "*MESSAGE*" \
130
+ "|CODE_LINE" \
131
+ "|*DOCUMENTATION*" \
132
+ "|TID" \
133
+ "|*_RAW" \
134
+ "|*_NSEC" \
135
+ "|*TIMESTAMP*" \
136
+ "|*_ID" \
137
+ "|*_ID_*" \
138
+ "|*_PID" \
139
+ "|*_TID" \
140
+ "|__*" \
141
+ ""
142
+
143
#define SYSTEMD_KEYS_INCLUDED_IN_FACETS \
43
- "_TRANSPORT" \
144
+ "_COMM" \
145
+ "|CONTAINER_NAME" \
146
+ "|CONTAINER_TAG" \
147
+ "|_TRANSPORT" \
148
"|SYSLOG_IDENTIFIER" \
149
"|SYSLOG_FACILITY" \
150
"|PRIORITY" \
47
- "|_UID" \
48
- "|_GID" \
151
"|_SYSTEMD_UNIT" \
152
"|_SYSTEMD_SLICE" \
51
- "|_COMM" \
153
+ "|_SYSTEMD_USER_UNIT" \
154
+ "|_SYSTEMD_USER_SLICE" \
155
+ "|_SYSTEMD_OWNER_UID" \
156
+ "|_UID" \
157
+ "|_GID" \
158
"|UNIT" \
53
- "|CONTAINER_NAME" \
159
+ "|USER_UNIT" \
160
"|IMAGE_NAME" \
161
+ "|ERRNO" \
162
+ "|_NAMESPACE" \
163
+ "|COREDUMP_COMM" \
164
+ "|COREDUMP_UNIT" \
165
+ "|COREDUMP_USER_UNIT" \
166
+ "|COREDUMP_SIGNAL_NAME" \
167
+ "|COREDUMP_CGROUP" \
168
+ "|_HOSTNAME" \
169
+ "|UNIT_RESULT" \
170
+ "|_RUNTIME_SCOPE" \
171
""
172
173
static netdata_mutex_t stdout_mutex = NETDATA_MUTEX_INITIALIZER;
175
176
// ----------------------------------------------------------------------------
177
62
-static inline sd_journal *netdata_open_systemd_journal(void) {
63
- sd_journal *j = NULL;
64
- int r;
65
-
66
- if(*netdata_configured_host_prefix) {
67
-#ifdef HAVE_SD_JOURNAL_OS_ROOT
68
- // Give our host prefix to systemd journal
69
- r = sd_journal_open_directory(&j, netdata_configured_host_prefix, SD_JOURNAL_OS_ROOT);
70
-#else
71
- char buf[FILENAME_MAX + 1];
72
- snprintfz(buf, FILENAME_MAX, "%s/var/log/journal", netdata_configured_host_prefix);
73
- r = sd_journal_open_directory(&j, buf, 0);
74
-#endif
75
- }
76
- else {
77
- // Open the system journal for reading
78
- r = sd_journal_open(&j, 0);
79
- }
80
-
81
- if (r < 0) {
82
- netdata_log_error("SYSTEMD-JOURNAL: Failed to open SystemD Journal, with error %d", r);
83
- return NULL;
84
- }
85
-
86
- return j;
87
-}
88
-
178
typedef enum {
179
+ ND_SD_JOURNAL_NO_FILE_MATCHED,
180
+ ND_SD_JOURNAL_FAILED_TO_OPEN,
181
ND_SD_JOURNAL_FAILED_TO_SEEK,
182
ND_SD_JOURNAL_TIMED_OUT,
183
ND_SD_JOURNAL_OK,
185
ND_SD_JOURNAL_CANCELLED,
186
} ND_SD_JOURNAL_STATUS;
187
188
+typedef enum {
189
+ SDJF_ALL = 0,
190
+ SDJF_LOCAL = (1 << 0),
191
+ SDJF_REMOTE = (1 << 1),
192
+ SDJF_SYSTEM = (1 << 2),
193
+ SDJF_USER = (1 << 3),
194
+ SDJF_NAMESPACE = (1 << 4),
195
+ SDJF_OTHER = (1 << 5),
196
+} SD_JOURNAL_FILE_SOURCE_TYPE;
197
+
198
+typedef struct function_query_status {
199
+ bool *cancelled; // a pointer to the cancelling boolean
200
+ usec_t stop_monotonic_ut;
201
+
202
+ usec_t started_monotonic_ut;
203
+
204
+ // request
205
+ SD_JOURNAL_FILE_SOURCE_TYPE source_type;
206
+ STRING *source;
207
+ usec_t after_ut;
208
+ usec_t before_ut;
209
+
210
+ struct {
211
+ usec_t start_ut;
212
+ usec_t stop_ut;
213
+ } anchor;
214
+
215
+ FACETS_ANCHOR_DIRECTION direction;
216
+ size_t entries;
217
+ usec_t if_modified_since;
218
+ bool delta;
219
+ bool tail;
220
+ bool data_only;
221
+ bool slice;
222
+ size_t filters;
223
+ usec_t last_modified;
224
+ const char *query;
225
+ const char *histogram;
226
+
227
+ // per file progress info
228
+ size_t cached_count;
229
+
230
+ // progress statistics
231
+ usec_t matches_setup_ut;
232
+ size_t rows_useful;
233
+ size_t rows_read;
234
+ size_t bytes_read;
235
+ size_t files_matched;
236
+ size_t file_working;
237
+} FUNCTION_QUERY_STATUS;
238
+
239
+struct journal_file {
240
+ STRING *source;
241
+ SD_JOURNAL_FILE_SOURCE_TYPE source_type;
242
+ usec_t file_last_modified_ut;
243
+ usec_t msg_first_ut;
244
+ usec_t msg_last_ut;
245
+ usec_t last_scan_ut;
246
+ size_t size;
247
+ bool logged_failure;
248
+ usec_t max_journal_vs_realtime_delta_ut;
249
+};
250
+
251
static inline bool netdata_systemd_journal_seek_to(sd_journal *j, usec_t timestamp) {
252
if(sd_journal_seek_realtime_usec(j, timestamp) < 0) {
253
netdata_log_error("SYSTEMD-JOURNAL: Failed to seek to %" PRIu64, timestamp);
260
return true;
261
}
262
109
-static inline void netdata_systemd_journal_process_row(sd_journal *j, FACETS *facets) {
263
+#define JD_SOURCE_REALTIME_TIMESTAMP "_SOURCE_REALTIME_TIMESTAMP"
264
+
265
+#define JOURNAL_VS_REALTIME_DELTA_DEFAULT_UT (2 * USEC_PER_SEC) // assume always 2 seconds latency
266
+#define JOURNAL_VS_REALTIME_DELTA_MAX_UT (2 * 60 * USEC_PER_SEC) // up to 2 minutes delta
267
+
268
+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) {
269
+ const char *k = data;
270
+ const char *equal = strchr(k, '=');
271
+ if(unlikely(!equal))
272
+ return false;
273
+
274
+ size_t kl = equal - k;
275
+
276
+ const char *v = ++equal;
277
+ size_t vl = data_length - kl - 1;
278
+
279
+ *key = k;
280
+ *key_length = kl;
281
+ *value = v;
282
+ *value_length = vl;
283
+
284
+ return true;
285
+}
286
+
287
+static inline size_t netdata_systemd_journal_process_row(sd_journal *j, FACETS *facets, struct journal_file *jf, usec_t *msg_ut) {
288
const void *data;
111
- size_t length;
289
+ size_t length, bytes = 0;
290
+
291
SD_JOURNAL_FOREACH_DATA(j, data, length) {
113
- const char *key = data;
114
- const char *equal = strchr(key, '=');
115
- if(unlikely(!equal))
116
- continue;
292
+ const char *key, *value;
293
+ size_t key_length, value_length;
294
118
- const char *value = ++equal;
119
- size_t key_length = value - key; // including '\0'
295
+ if(!parse_journal_field(data, length, &key, &key_length, &value, &value_length))
296
+ continue;
297
121
- char key_copy[key_length];
122
- memcpy(key_copy, key, key_length - 1);
123
- key_copy[key_length - 1] = '\0';
298
+ usec_t origin_journal_ut = *msg_ut;
299
+
300
+ if(unlikely(key_length == sizeof(JD_SOURCE_REALTIME_TIMESTAMP) - 1 &&
301
+ memcmp(key, JD_SOURCE_REALTIME_TIMESTAMP, sizeof(JD_SOURCE_REALTIME_TIMESTAMP) - 1) == 0)) {
302
+ usec_t ut = str2ull(value, NULL);
303
+ if(ut && ut < *msg_ut) {
304
+ usec_t delta = *msg_ut - ut;
305
+ *msg_ut = ut;
306
+
307
+ if(delta > JOURNAL_VS_REALTIME_DELTA_MAX_UT)
308
+ delta = JOURNAL_VS_REALTIME_DELTA_MAX_UT;
309
+
310
+ // update max_journal_vs_realtime_delta_ut if the delta increased
311
+ usec_t expected = jf->max_journal_vs_realtime_delta_ut;
312
+ do {
313
+ if(delta <= expected)
314
+ break;
315
+ } while(!__atomic_compare_exchange_n(&jf->max_journal_vs_realtime_delta_ut, &expected, delta, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
316
+
317
+ internal_error(delta > expected,
318
+ "increased max_journal_vs_realtime_delta_ut from %"PRIu64" to %"PRIu64", "
319
+ "journal %"PRIu64", actual %"PRIu64" (delta %"PRIu64")"
320
+ , expected, delta, origin_journal_ut, *msg_ut, origin_journal_ut - (*msg_ut));
321
+ }
322
+ }
323
125
- size_t value_length = length - key_length; // without '\0'
126
- facets_add_key_value_length(facets, key_copy, key_length - 1, value, value_length <= FACET_MAX_VALUE_LENGTH ? value_length : FACET_MAX_VALUE_LENGTH);
324
+ bytes += length;
325
+ facets_add_key_value_length(facets, key, key_length, value, value_length <= FACET_MAX_VALUE_LENGTH ? value_length : FACET_MAX_VALUE_LENGTH);
326
}
327
+
328
+ return bytes;
329
}
330
130
-static inline ND_SD_JOURNAL_STATUS check_stop(size_t row_counter, const bool *cancelled, usec_t stop_monotonic_ut) {
131
- if((row_counter % 1000) == 0) {
132
- if(cancelled && __atomic_load_n(cancelled, __ATOMIC_RELAXED)) {
133
- internal_error(true, "Function has been cancelled");
134
- return ND_SD_JOURNAL_CANCELLED;
135
- }
331
+#define FUNCTION_PROGRESS_UPDATE_ROWS(rows_read, rows) __atomic_fetch_add(&(rows_read), rows, __ATOMIC_RELAXED)
332
+#define FUNCTION_PROGRESS_UPDATE_BYTES(bytes_read, bytes) __atomic_fetch_add(&(bytes_read), bytes, __ATOMIC_RELAXED)
333
+#define FUNCTION_PROGRESS_EVERY_ROWS 10000
334
137
- if(now_monotonic_usec() > stop_monotonic_ut) {
138
- internal_error(true, "Function timed out");
139
- return ND_SD_JOURNAL_TIMED_OUT;
140
- }
335
+static inline ND_SD_JOURNAL_STATUS check_stop(const bool *cancelled, const usec_t *stop_monotonic_ut) {
336
+ if(cancelled && __atomic_load_n(cancelled, __ATOMIC_RELAXED)) {
337
+ internal_error(true, "Function has been cancelled");
338
+ return ND_SD_JOURNAL_CANCELLED;
339
+ }
340
+
341
+ if(now_monotonic_usec() > __atomic_load_n(stop_monotonic_ut, __ATOMIC_RELAXED)) {
342
+ internal_error(true, "Function timed out");
343
+ return ND_SD_JOURNAL_TIMED_OUT;
344
}
345
346
return ND_SD_JOURNAL_OK;
347
}
348
146
-ND_SD_JOURNAL_STATUS netdata_systemd_journal_query_full(
349
+ND_SD_JOURNAL_STATUS netdata_systemd_journal_query_backward(
350
sd_journal *j, BUFFER *wb __maybe_unused, FACETS *facets,
148
- usec_t after_ut, usec_t before_ut,
149
- usec_t if_modified_since, usec_t stop_monotonic_ut, usec_t *last_modified,
150
- bool *cancelled) {
151
- if(!netdata_systemd_journal_seek_to(j, before_ut))
351
+ struct journal_file *jf, FUNCTION_QUERY_STATUS *fqs) {
352
+
353
+ usec_t anchor_delta = __atomic_load_n(&jf->max_journal_vs_realtime_delta_ut, __ATOMIC_RELAXED);
354
+
355
+ usec_t start_ut = ((fqs->data_only && fqs->anchor.start_ut) ? fqs->anchor.start_ut : fqs->before_ut) + anchor_delta;
356
+ usec_t stop_ut = (fqs->data_only && fqs->anchor.stop_ut) ? fqs->anchor.stop_ut : fqs->after_ut;
357
+
358
+ if(!netdata_systemd_journal_seek_to(j, start_ut))
359
return ND_SD_JOURNAL_FAILED_TO_SEEK;
360
361
size_t errors_no_timestamp = 0;
155
- usec_t first_msg_ut = 0;
156
- size_t row_counter = 0;
157
-
158
- // the entries are not guaranteed to be sorted, so we process up to 100 entries beyond
159
- // the end of the query to find possibly useful logs for our time-frame
160
- size_t excess_rows_allowed = SYSTEMD_JOURNAL_EXCESS_ROWS_ALLOWED;
362
+ usec_t earliest_msg_ut = 0;
363
+ size_t row_counter = 0, last_row_counter = 0;
364
+ size_t bytes = 0, last_bytes = 0;
365
366
ND_SD_JOURNAL_STATUS status = ND_SD_JOURNAL_OK;
367
368
facets_rows_begin(facets);
369
while (status == ND_SD_JOURNAL_OK && sd_journal_previous(j) > 0) {
166
- row_counter++;
167
-
168
- usec_t msg_ut;
169
- if(sd_journal_get_realtime_usec(j, &msg_ut) < 0) {
370
+ usec_t msg_ut = 0;
371
+ if(sd_journal_get_realtime_usec(j, &msg_ut) < 0 || !msg_ut) {
372
errors_no_timestamp++;
373
continue;
374
}
375
174
- if(unlikely(!first_msg_ut)) {
175
- if(msg_ut == if_modified_since) {
176
- return ND_SD_JOURNAL_NOT_MODIFIED;
177
- }
178
-
179
- first_msg_ut = msg_ut;
180
- }
376
+ if(unlikely(msg_ut > earliest_msg_ut))
377
+ earliest_msg_ut = msg_ut;
378
182
- if (msg_ut > before_ut)
379
+ if (unlikely(msg_ut > start_ut))
380
continue;
381
185
- if (msg_ut < after_ut) {
186
- if(--excess_rows_allowed == 0)
187
- break;
382
+ if (unlikely(msg_ut < stop_ut))
383
+ break;
384
189
- continue;
385
+ bytes += netdata_systemd_journal_process_row(j, facets, jf, &msg_ut);
386
+ if(facets_row_finished(facets, msg_ut))
387
+ fqs->rows_useful++;
388
+
389
+ row_counter++;
390
+ if(row_counter % 100 == 0 && fqs->data_only && facets_rows(facets) >= fqs->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
192
- netdata_systemd_journal_process_row(j, facets);
193
- facets_row_finished(facets, msg_ut);
397
+ if(row_counter % FUNCTION_PROGRESS_EVERY_ROWS == 0) {
398
+ FUNCTION_PROGRESS_UPDATE_ROWS(fqs->rows_read, row_counter - last_row_counter);
399
+ last_row_counter = row_counter;
400
+
401
+ FUNCTION_PROGRESS_UPDATE_BYTES(fqs->bytes_read, bytes - last_bytes);
402
+ last_bytes = bytes;
403
195
- status = check_stop(row_counter, cancelled, stop_monotonic_ut);
404
+ status = check_stop(fqs->cancelled, &fqs->stop_monotonic_ut);
405
+ }
406
}
407
408
+ FUNCTION_PROGRESS_UPDATE_ROWS(fqs->rows_read, row_counter - last_row_counter);
409
+ FUNCTION_PROGRESS_UPDATE_BYTES(fqs->bytes_read, bytes - last_bytes);
410
+
411
if(errors_no_timestamp)
412
netdata_log_error("SYSTEMD-JOURNAL: %zu lines did not have timestamps", errors_no_timestamp);
413
201
- *last_modified = first_msg_ut;
414
+ if(earliest_msg_ut > fqs->last_modified)
415
+ fqs->last_modified = earliest_msg_ut;
416
417
return status;
418
}
419
206
-ND_SD_JOURNAL_STATUS netdata_systemd_journal_query_data_forward(
420
+ND_SD_JOURNAL_STATUS netdata_systemd_journal_query_forward(
421
sd_journal *j, BUFFER *wb __maybe_unused, FACETS *facets,
208
- usec_t after_ut, usec_t before_ut,
209
- usec_t anchor, size_t entries, usec_t stop_monotonic_ut,
210
- bool *cancelled) {
422
+ struct journal_file *jf, FUNCTION_QUERY_STATUS *fqs) {
423
+
424
+ usec_t anchor_delta = __atomic_load_n(&jf->max_journal_vs_realtime_delta_ut, __ATOMIC_RELAXED);
425
212
- if(!netdata_systemd_journal_seek_to(j, anchor))
426
+ usec_t start_ut = (fqs->data_only && fqs->anchor.start_ut) ? fqs->anchor.start_ut : fqs->after_ut;
427
+ usec_t stop_ut = ((fqs->data_only && fqs->anchor.stop_ut) ? fqs->anchor.stop_ut : fqs->before_ut) + anchor_delta;
428
+
429
+ if(!netdata_systemd_journal_seek_to(j, start_ut))
430
return ND_SD_JOURNAL_FAILED_TO_SEEK;
431
432
size_t errors_no_timestamp = 0;
216
- size_t row_counter = 0;
217
- size_t rows_added = 0;
218
-
219
- // the entries are not guaranteed to be sorted, so we process up to 100 entries beyond
220
- // the end of the query to find possibly useful logs for our time-frame
221
- size_t excess_rows_allowed = SYSTEMD_JOURNAL_EXCESS_ROWS_ALLOWED;
433
+ usec_t earliest_msg_ut = 0;
434
+ size_t row_counter = 0, last_row_counter = 0;
435
+ size_t bytes = 0, last_bytes = 0;
436
437
ND_SD_JOURNAL_STATUS status = ND_SD_JOURNAL_OK;
438
439
facets_rows_begin(facets);
440
while (status == ND_SD_JOURNAL_OK && sd_journal_next(j) > 0) {
227
- row_counter++;
228
-
229
- usec_t msg_ut;
230
- if(sd_journal_get_realtime_usec(j, &msg_ut) < 0) {
441
+ usec_t msg_ut = 0;
442
+ if(sd_journal_get_realtime_usec(j, &msg_ut) < 0 || !msg_ut) {
443
errors_no_timestamp++;
444
continue;
445
}
446
235
- if (msg_ut > before_ut || msg_ut <= anchor)
447
+ if(likely(msg_ut > earliest_msg_ut))
448
+ earliest_msg_ut = msg_ut;
449
+
450
+ if (unlikely(msg_ut < start_ut))
451
continue;
452
238
- if (msg_ut < after_ut) {
239
- if(--excess_rows_allowed == 0)
240
- break;
453
+ if (unlikely(msg_ut > stop_ut))
454
+ break;
455
242
- continue;
456
+ bytes += netdata_systemd_journal_process_row(j, facets, jf, &msg_ut);
457
+ if(facets_row_finished(facets, msg_ut))
458
+ fqs->rows_useful++;
459
+
460
+ row_counter++;
461
+ if(row_counter % 100 == 0 && fqs->data_only && facets_rows(facets) >= fqs->entries) {
462
+ usec_t newest = facets_row_newest_ut(facets);
463
+ if(newest && msg_ut > (newest + anchor_delta))
464
+ break;
465
}
466
245
- if(rows_added > entries && --excess_rows_allowed == 0)
246
- break;
467
+ if(row_counter % FUNCTION_PROGRESS_EVERY_ROWS == 0) {
468
+ FUNCTION_PROGRESS_UPDATE_ROWS(fqs->rows_read, row_counter - last_row_counter);
469
+ last_row_counter = row_counter;
470
248
- netdata_systemd_journal_process_row(j, facets);
249
- facets_row_finished(facets, msg_ut);
250
- rows_added++;
471
+ FUNCTION_PROGRESS_UPDATE_BYTES(fqs->bytes_read, bytes - last_bytes);
472
+ last_bytes = bytes;
473
252
- status = check_stop(row_counter, cancelled, stop_monotonic_ut);
474
+ status = check_stop(fqs->cancelled, &fqs->stop_monotonic_ut);
475
+ }
476
}
477
478
+ FUNCTION_PROGRESS_UPDATE_ROWS(fqs->rows_read, row_counter - last_row_counter);
479
+ FUNCTION_PROGRESS_UPDATE_BYTES(fqs->bytes_read, bytes - last_bytes);
480
+
481
if(errors_no_timestamp)
482
netdata_log_error("SYSTEMD-JOURNAL: %zu lines did not have timestamps", errors_no_timestamp);
483
484
+ if(earliest_msg_ut > fqs->last_modified)
485
+ fqs->last_modified = earliest_msg_ut;
486
+
487
return status;
488
}
489
261
-ND_SD_JOURNAL_STATUS netdata_systemd_journal_query_data_backward(
262
- sd_journal *j, BUFFER *wb __maybe_unused, FACETS *facets,
263
- usec_t after_ut, usec_t before_ut,
264
- usec_t anchor, size_t entries, usec_t stop_monotonic_ut,
265
- bool *cancelled) {
490
+bool netdata_systemd_journal_check_if_modified_since(sd_journal *j, usec_t seek_to, usec_t last_modified) {
491
+ // return true, if data have been modified since the timestamp
492
267
- if(!netdata_systemd_journal_seek_to(j, anchor))
268
- return ND_SD_JOURNAL_FAILED_TO_SEEK;
493
+ if(!last_modified || !seek_to)
494
+ return false;
495
+
496
+ if(!netdata_systemd_journal_seek_to(j, seek_to))
497
+ return false;
498
+
499
+ usec_t first_msg_ut = 0;
500
+ while (sd_journal_previous(j) > 0) {
501
+ usec_t msg_ut;
502
+ if(sd_journal_get_realtime_usec(j, &msg_ut) < 0)
503
+ continue;
504
+
505
+ first_msg_ut = msg_ut;
506
+ break;
507
+ }
508
+
509
+ return first_msg_ut != last_modified;
510
+}
511
+
512
+#ifdef HAVE_SD_JOURNAL_RESTART_FIELDS
513
+static bool netdata_systemd_filtering_by_journal(sd_journal *j, FACETS *facets, FUNCTION_QUERY_STATUS *fqs) {
514
+ const char *field = NULL;
515
+ const void *data = NULL;
516
+ size_t data_length;
517
+ size_t added_keys = 0;
518
+ size_t failures = 0;
519
+ size_t filters_added = 0;
520
+
521
+ SD_JOURNAL_FOREACH_FIELD(j, field) {
522
+ bool interesting;
523
+
524
+ if(fqs->data_only)
525
+ interesting = facets_key_name_is_filter(facets, field);
526
+ else
527
+ interesting = facets_key_name_is_facet(facets, field);
528
+
529
+ if(interesting) {
530
+ if(sd_journal_query_unique(j, field) >= 0) {
531
+ bool added_this_key = false;
532
+ size_t added_values = 0;
533
+
534
+ SD_JOURNAL_FOREACH_UNIQUE(j, data, data_length) {
535
+ const char *key, *value;
536
+ size_t key_length, value_length;
537
+
538
+ if(!parse_journal_field(data, data_length, &key, &key_length, &value, &value_length))
539
+ continue;
540
+
541
+ facets_add_possible_value_name_to_key(facets, key, key_length, value, value_length);
542
+
543
+ if(!facets_key_name_value_length_is_selected(facets, key, key_length, value, value_length))
544
+ continue;
545
+
546
+ if(added_keys && !added_this_key) {
547
+ if(sd_journal_add_conjunction(j) < 0)
548
+ failures++;
549
+
550
+ added_this_key = true;
551
+ added_keys++;
552
+ }
553
+ else if(added_values)
554
+ if(sd_journal_add_disjunction(j) < 0)
555
+ failures++;
556
+
557
+ if(sd_journal_add_match(j, data, data_length) < 0)
558
+ failures++;
559
+
560
+ added_values++;
561
+ filters_added++;
562
+ }
563
+ }
564
+ }
565
+ }
566
+
567
+ if(failures) {
568
+ netdata_log_error("failed to setup journal filter, will run the full query.");
569
+ sd_journal_flush_matches(j);
570
+ return true;
571
+ }
572
+
573
+ return filters_added ? true : false;
574
+}
575
+#endif // HAVE_SD_JOURNAL_RESTART_FIELDS
576
+
577
+static ND_SD_JOURNAL_STATUS netdata_systemd_journal_query_one_file(
578
+ const char *filename, BUFFER *wb, FACETS *facets,
579
+ struct journal_file *jf, FUNCTION_QUERY_STATUS *fqs) {
580
+
581
+ sd_journal *j = NULL;
582
+ errno = 0;
583
+
584
+#ifdef HAVE_SD_JOURNAL_OPEN_FILES_FD
585
+ int fd = open(filename, O_RDONLY);
586
+ fstat_cache_enable(fd);
587
+
588
+ if(sd_journal_open_files_fd(&j, &fd, 1, ND_SD_JOURNAL_OPEN_FLAGS) < 0 || !j) {
589
+ fqs->cached_count += fstat_cache_disable(fd);
590
+ close(fd);
591
+ return ND_SD_JOURNAL_FAILED_TO_OPEN;
592
+ }
593
+#else // !HAVE_SD_JOURNAL_OPEN_FILES_FD
594
+
595
+ const char *paths[2] = {
596
+ [0] = filename,
597
+ [1] = NULL,
598
+ };
599
+ if(sd_journal_open_files(&j, paths, ND_SD_JOURNAL_OPEN_FLAGS) < 0 || !j)
600
+ return ND_SD_JOURNAL_FAILED_TO_OPEN;
601
+
602
+#endif // !HAVE_SD_JOURNAL_OPEN_FILES_FD
603
+
604
+ ND_SD_JOURNAL_STATUS status;
605
+ bool matches_filters = true;
606
+
607
+#ifdef HAVE_SD_JOURNAL_RESTART_FIELDS
608
+ if(fqs->slice) {
609
+ usec_t started = now_monotonic_usec();
610
+
611
+ matches_filters = netdata_systemd_filtering_by_journal(j, facets, fqs) || !fqs->filters;
612
+ usec_t ended = now_monotonic_usec();
613
+
614
+ fqs->matches_setup_ut += (ended - started);
615
+ }
616
+#endif // HAVE_SD_JOURNAL_RESTART_FIELDS
617
+
618
+ if(matches_filters) {
619
+ if(fqs->direction == FACETS_ANCHOR_DIRECTION_FORWARD)
620
+ status = netdata_systemd_journal_query_forward(j, wb, facets, jf, fqs);
621
+ else
622
+ status = netdata_systemd_journal_query_backward(j, wb, facets, jf, fqs);
623
+ }
624
+ else
625
+ status = ND_SD_JOURNAL_NO_FILE_MATCHED;
626
+
627
+ sd_journal_close(j);
628
+
629
+#ifdef HAVE_SD_JOURNAL_OPEN_FILES_FD
630
+ fqs->cached_count += fstat_cache_disable(fd);
631
+ close(fd);
632
+#endif
633
+
634
+ return status;
635
+}
636
+
637
+// ----------------------------------------------------------------------------
638
+// journal files registry
639
+
640
+#define VAR_LOG_JOURNAL_MAX_DEPTH 10
641
+#define MAX_JOURNAL_DIRECTORIES 100
642
+
643
+struct journal_directory {
644
+ char *path;
645
+ bool logged_failure;
646
+};
647
+
648
+static struct journal_directory journal_directories[MAX_JOURNAL_DIRECTORIES] = { 0 };
649
+static DICTIONARY *journal_files_registry = NULL;
650
+static DICTIONARY *used_hashes_registry = NULL;
651
+
652
+static usec_t systemd_journal_session = 0;
653
+
654
+static void buffer_json_journal_versions(BUFFER *wb) {
655
+ buffer_json_member_add_object(wb, "versions");
656
+ {
657
+ buffer_json_member_add_uint64(wb, "sources",
658
+ systemd_journal_session + dictionary_version(journal_files_registry));
659
+ }
660
+ buffer_json_object_close(wb);
661
+}
662
+
663
+static void journal_file_update_msg_ut(const char *filename, struct journal_file *jf) {
664
+ const char *files[2] = {
665
+ [0] = filename,
666
+ [1] = NULL,
667
+ };
668
+
669
+ sd_journal *j = NULL;
670
+ if(sd_journal_open_files(&j, files, ND_SD_JOURNAL_OPEN_FLAGS) < 0 || !j) {
671
+ if(!jf->logged_failure) {
672
+ netdata_log_error("cannot open journal file '%s', using file timestamps to understand time-frame.", filename);
673
+ jf->logged_failure = true;
674
+ }
675
+
676
+ jf->msg_first_ut = 0;
677
+ jf->msg_last_ut = jf->file_last_modified_ut;
678
+ return;
679
+ }
680
+
681
+ usec_t first_ut = 0, last_ut = 0;
682
+
683
+ if(sd_journal_seek_head(j) < 0 || sd_journal_next(j) < 0 || sd_journal_get_realtime_usec(j, &first_ut) < 0 || !first_ut) {
684
+ internal_error(true, "cannot find the timestamp of the first message in '%s'", filename);
685
+ first_ut = 0;
686
+ }
687
+
688
+ if(sd_journal_seek_tail(j) < 0 || sd_journal_previous(j) < 0 || sd_journal_get_realtime_usec(j, &last_ut) < 0 || !last_ut) {
689
+ internal_error(true, "cannot find the timestamp of the last message in '%s'", filename);
690
+ last_ut = jf->file_last_modified_ut;
691
+ }
692
+
693
+ sd_journal_close(j);
694
+
695
+ if(first_ut > last_ut) {
696
+ internal_error(true, "timestamps are flipped in file '%s'", filename);
697
+ usec_t t = first_ut;
698
+ first_ut = last_ut;
699
+ last_ut = t;
700
+ }
701
+
702
+ jf->msg_first_ut = first_ut;
703
+ jf->msg_last_ut = last_ut;
704
+}
705
+
706
+static STRING *string_strdupz_source(const char *s, const char *e, size_t max_len, const char *prefix) {
707
+ char buf[max_len];
708
+ size_t len;
709
+ char *dst = buf;
710
+
711
+ if(prefix) {
712
+ len = strlen(prefix);
713
+ memcpy(buf, prefix, len);
714
+ dst = &buf[len];
715
+ max_len -= len;
716
+ }
717
+
718
+ len = e - s;
719
+ if(len >= max_len)
720
+ len = max_len - 1;
721
+ memcpy(dst, s, len);
722
+ dst[len] = '\0';
723
+ buf[max_len - 1] = '\0';
724
+
725
+ for(size_t i = 0; buf[i] ;i++)
726
+ if(!isalnum(buf[i]) && buf[i] != '-' && buf[i] != '.' && buf[i] != ':')
727
+ buf[i] = '_';
728
+
729
+ return string_strdupz(buf);
730
+}
731
+
732
+static void files_registry_insert_cb(const DICTIONARY_ITEM *item, void *value, void *data __maybe_unused) {
733
+ struct journal_file *jf = value;
734
+ const char *filename = dictionary_acquired_item_name(item);
735
+
736
+ // based on the filename
737
+ // decide the source to show to the user
738
+ const char *s = strrchr(filename, '/');
739
+ if(s) {
740
+ if(strstr(filename, "/remote/"))
741
+ jf->source_type = SDJF_REMOTE;
742
+ else {
743
+ const char *t = s - 1;
744
+ while(t >= filename && *t != '.' && *t != '/')
745
+ t--;
746
+
747
+ if(t >= filename && *t == '.') {
748
+ jf->source_type = SDJF_NAMESPACE;
749
+ jf->source = string_strdupz_source(t + 1, s, SYSTEMD_JOURNAL_MAX_SOURCE_LEN, "namespace-");
750
+ }
751
+ else
752
+ jf->source_type = SDJF_LOCAL;
753
+ }
754
+
755
+ if(strncmp(s, "/system", 7) == 0)
756
+ jf->source_type |= SDJF_SYSTEM;
757
+
758
+ else if(strncmp(s, "/user", 5) == 0)
759
+ jf->source_type |= SDJF_USER;
760
+
761
+ else if(strncmp(s, "/remote-", 8) == 0) {
762
+ jf->source_type |= SDJF_REMOTE;
763
+
764
+ s = &s[8]; // skip "/remote-"
765
+
766
+ char *e = strchr(s, '@');
767
+ if(!e)
768
+ e = strstr(s, ".journal");
769
+
770
+ if(e) {
771
+ const char *d = s;
772
+ for(; d < e && (isdigit(*d) || *d == '.' || *d == ':') ; d++) ;
773
+ if(d == e) {
774
+ // a valid IP address
775
+ char ip[e - s + 1];
776
+ memcpy(ip, s, e - s);
777
+ ip[e - s] = '\0';
778
+ char buf[SYSTEMD_JOURNAL_MAX_SOURCE_LEN];
779
+ if(ip_to_hostname(ip, buf, sizeof(buf)))
780
+ jf->source = string_strdupz_source(buf, &buf[strlen(buf)], SYSTEMD_JOURNAL_MAX_SOURCE_LEN, "remote-");
781
+ else {
782
+ internal_error(true, "Cannot find the hostname for IP '%s'", ip);
783
+ jf->source = string_strdupz_source(s, e, SYSTEMD_JOURNAL_MAX_SOURCE_LEN, "remote-");
784
+ }
785
+ }
786
+ else
787
+ jf->source = string_strdupz_source(s, e, SYSTEMD_JOURNAL_MAX_SOURCE_LEN, "remote-");
788
+ }
789
+ else
790
+ jf->source_type |= SDJF_OTHER;
791
+ }
792
+ else
793
+ jf->source_type |= SDJF_OTHER;
794
+ }
795
+ else
796
+ jf->source_type = SDJF_LOCAL | SDJF_OTHER;
797
+
798
+ journal_file_update_msg_ut(filename, jf);
799
+
800
+ internal_error(true,
801
+ "found journal file '%s', type %d, source '%s', "
802
+ "file modified: %"PRIu64", "
803
+ "msg {first: %"PRIu64", last: %"PRIu64"}",
804
+ filename, jf->source_type, jf->source ? string2str(jf->source) : "<unset>",
805
+ jf->file_last_modified_ut,
806
+ jf->msg_first_ut, jf->msg_last_ut);
807
+}
808
+
809
+static bool files_registry_conflict_cb(const DICTIONARY_ITEM *item, void *old_value, void *new_value, void *data __maybe_unused) {
810
+ struct journal_file *jf = old_value;
811
+ struct journal_file *njf = new_value;
812
+
813
+ if(njf->last_scan_ut > jf->last_scan_ut)
814
+ jf->last_scan_ut = njf->last_scan_ut;
815
+
816
+ if(njf->file_last_modified_ut > jf->file_last_modified_ut) {
817
+ jf->file_last_modified_ut = njf->file_last_modified_ut;
818
+ jf->size = njf->size;
819
+
820
+ const char *filename = dictionary_acquired_item_name(item);
821
+ journal_file_update_msg_ut(filename, jf);
822
+
823
+// internal_error(true,
824
+// "updated journal file '%s', type %d, "
825
+// "file modified: %"PRIu64", "
826
+// "msg {first: %"PRIu64", last: %"PRIu64"}",
827
+// filename, jf->source_type,
828
+// jf->file_last_modified_ut,
829
+// jf->msg_first_ut, jf->msg_last_ut);
830
+ }
831
+
832
+ return false;
833
+}
834
+
835
+#define SDJF_SOURCE_ALL_NAME "all"
836
+#define SDJF_SOURCE_LOCAL_NAME "all-local-logs"
837
+#define SDJF_SOURCE_LOCAL_SYSTEM_NAME "all-local-system-logs"
838
+#define SDJF_SOURCE_LOCAL_USERS_NAME "all-local-user-logs"
839
+#define SDJF_SOURCE_LOCAL_OTHER_NAME "all-uncategorized"
840
+#define SDJF_SOURCE_NAMESPACES_NAME "all-local-namespaces"
841
+#define SDJF_SOURCE_REMOTES_NAME "all-remote-systems"
842
+
843
+struct journal_file_source {
844
+ usec_t first_ut;
845
+ usec_t last_ut;
846
+ size_t count;
847
+ uint64_t size;
848
+};
849
+
850
+static void human_readable_size_ib(uint64_t size, char *dst, size_t dst_len) {
851
+ if(size > 1024ULL * 1024 * 1024 * 1024)
852
+ snprintfz(dst, dst_len, "%0.2f TiB", (double)size / 1024.0 / 1024.0 / 1024.0 / 1024.0);
853
+ else if(size > 1024ULL * 1024 * 1024)
854
+ snprintfz(dst, dst_len, "%0.2f GiB", (double)size / 1024.0 / 1024.0 / 1024.0);
855
+ else if(size > 1024ULL * 1024)
856
+ snprintfz(dst, dst_len, "%0.2f MiB", (double)size / 1024.0 / 1024.0);
857
+ else if(size > 1024ULL)
858
+ snprintfz(dst, dst_len, "%0.2f KiB", (double)size / 1024.0);
859
+ else
860
+ snprintfz(dst, dst_len, "%"PRIu64" B", size);
861
+}
862
+
863
+#define print_duration(dst, dst_len, pos, remaining, duration, one, many, printed) do { \
864
+ if((remaining) > (duration)) { \
865
+ uint64_t _count = (remaining) / (duration); \
866
+ uint64_t _rem = (remaining) - (_count * (duration)); \
867
+ (pos) += snprintfz(&(dst)[pos], (dst_len) - (pos), "%s%s%"PRIu64" %s", (printed) ? ", " : "", _rem ? "" : "and ", _count, _count > 1 ? (many) : (one)); \
868
+ (remaining) = _rem; \
869
+ (printed) = true; \
870
+ } \
871
+} while(0)
872
+
873
+static void human_readable_duration_s(time_t duration_s, char *dst, size_t dst_len) {
874
+ if(duration_s < 0)
875
+ duration_s = -duration_s;
876
+
877
+ size_t pos = 0;
878
+ dst[0] = 0 ;
879
+
880
+ bool printed = false;
881
+ print_duration(dst, dst_len, pos, duration_s, 86400 * 365, "year", "years", printed);
882
+ print_duration(dst, dst_len, pos, duration_s, 86400 * 30, "month", "months", printed);
883
+ print_duration(dst, dst_len, pos, duration_s, 86400 * 1, "day", "days", printed);
884
+ print_duration(dst, dst_len, pos, duration_s, 3600 * 1, "hour", "hours", printed);
885
+ print_duration(dst, dst_len, pos, duration_s, 60 * 1, "min", "mins", printed);
886
+ print_duration(dst, dst_len, pos, duration_s, 1, "sec", "secs", printed);
887
+}
888
+
889
+static int journal_file_to_json_array_cb(const DICTIONARY_ITEM *item, void *entry, void *data) {
890
+ struct journal_file_source *jfs = entry;
891
+ BUFFER *wb = data;
892
+
893
+ const char *name = dictionary_acquired_item_name(item);
894
+
895
+ buffer_json_add_array_item_object(wb);
896
+ {
897
+ char size_for_humans[100];
898
+ human_readable_size_ib(jfs->size, size_for_humans, sizeof(size_for_humans));
899
+
900
+ char duration_for_humans[1024];
901
+ human_readable_duration_s((time_t)((jfs->last_ut - jfs->first_ut) / USEC_PER_SEC),
902
+ duration_for_humans, sizeof(duration_for_humans));
903
+
904
+ char info[1024];
905
+ snprintfz(info, sizeof(info), "%zu files, with a total size of %s, covering %s",
906
+ jfs->count, size_for_humans, duration_for_humans);
907
+
908
+ buffer_json_member_add_string(wb, "id", name);
909
+ buffer_json_member_add_string(wb, "name", name);
910
+ buffer_json_member_add_string(wb, "pill", size_for_humans);
911
+ buffer_json_member_add_string(wb, "info", info);
912
+ }
913
+ buffer_json_object_close(wb); // options object
914
+
915
+ return 1;
916
+}
917
+
918
+static bool journal_file_merge_sizes(const DICTIONARY_ITEM *item __maybe_unused, void *old_value, void *new_value , void *data __maybe_unused) {
919
+ struct journal_file_source *jfs = old_value, *njfs = new_value;
920
+ jfs->count += njfs->count;
921
+ jfs->size += njfs->size;
922
+
923
+ if(njfs->first_ut && njfs->first_ut < jfs->first_ut)
924
+ jfs->first_ut = njfs->first_ut;
925
+
926
+ if(njfs->last_ut && njfs->last_ut > jfs->last_ut)
927
+ jfs->last_ut = njfs->last_ut;
928
+
929
+ return false;
930
+}
931
+
932
+static void available_journal_file_sources_to_json_array(BUFFER *wb) {
933
+ DICTIONARY *dict = dictionary_create(DICT_OPTION_SINGLE_THREADED|DICT_OPTION_NAME_LINK_DONT_CLONE|DICT_OPTION_DONT_OVERWRITE_VALUE);
934
+ dictionary_register_conflict_callback(dict, journal_file_merge_sizes, NULL);
935
+
936
+ struct journal_file_source t = { 0 };
937
+
938
+ struct journal_file *jf;
939
+ dfe_start_read(journal_files_registry, jf) {
940
+ t.first_ut = jf->msg_first_ut;
941
+ t.last_ut = jf->msg_last_ut;
942
+ t.count = 1;
943
+ t.size = jf->size;
944
+
945
+ dictionary_set(dict, SDJF_SOURCE_ALL_NAME, &t, sizeof(t));
946
+
947
+ if((jf->source_type & (SDJF_LOCAL)) == (SDJF_LOCAL))
948
+ dictionary_set(dict, SDJF_SOURCE_LOCAL_NAME, &t, sizeof(t));
949
+ if((jf->source_type & (SDJF_LOCAL | SDJF_SYSTEM)) == (SDJF_LOCAL | SDJF_SYSTEM))
950
+ dictionary_set(dict, SDJF_SOURCE_LOCAL_SYSTEM_NAME, &t, sizeof(t));
951
+ if((jf->source_type & (SDJF_LOCAL | SDJF_USER)) == (SDJF_LOCAL | SDJF_USER))
952
+ dictionary_set(dict, SDJF_SOURCE_LOCAL_USERS_NAME, &t, sizeof(t));
953
+ if((jf->source_type & (SDJF_LOCAL | SDJF_OTHER)) == (SDJF_LOCAL | SDJF_OTHER))
954
+ dictionary_set(dict, SDJF_SOURCE_LOCAL_OTHER_NAME, &t, sizeof(t));
955
+ if((jf->source_type & (SDJF_NAMESPACE)) == (SDJF_NAMESPACE))
956
+ dictionary_set(dict, SDJF_SOURCE_NAMESPACES_NAME, &t, sizeof(t));
957
+ if((jf->source_type & (SDJF_REMOTE)) == (SDJF_REMOTE))
958
+ dictionary_set(dict, SDJF_SOURCE_REMOTES_NAME, &t, sizeof(t));
959
+ if(jf->source)
960
+ dictionary_set(dict, string2str(jf->source), &t, sizeof(t));
961
+ }
962
+ dfe_done(jf);
963
+
964
+ dictionary_sorted_walkthrough_read(dict, journal_file_to_json_array_cb, wb);
965
+
966
+ dictionary_destroy(dict);
967
+}
968
+
969
+static void files_registry_delete_cb(const DICTIONARY_ITEM *item, void *value, void *data __maybe_unused) {
970
+ struct journal_file *jf = value; (void)jf;
971
+ const char *filename = dictionary_acquired_item_name(item); (void)filename;
972
+
973
+ string_freez(jf->source);
974
+ internal_error(true, "removed journal file '%s'", filename);
975
+}
976
+
977
+void journal_directory_scan(const char *dirname, int depth, usec_t last_scan_ut) {
978
+ static const char *ext = ".journal";
979
+ static const size_t ext_len = sizeof(".journal") - 1;
980
+
981
+ if (depth > VAR_LOG_JOURNAL_MAX_DEPTH)
982
+ return;
983
+
984
+ DIR *dir;
985
+ struct dirent *entry;
986
+ struct stat info;
987
+ char absolute_path[FILENAME_MAX];
988
+
989
+ // Open the directory.
990
+ if ((dir = opendir(dirname)) == NULL) {
991
+ if(errno != ENOENT && errno != ENOTDIR)
992
+ netdata_log_error("Cannot opendir() '%s'", dirname);
993
+ return;
994
+ }
995
+
996
+ // Read each entry in the directory.
997
+ while ((entry = readdir(dir)) != NULL) {
998
+ snprintfz(absolute_path, sizeof(absolute_path), "%s/%s", dirname, entry->d_name);
999
+ if (stat(absolute_path, &info) != 0) {
1000
+ netdata_log_error("Failed to stat() '%s", absolute_path);
1001
+ continue;
1002
+ }
1003
+
1004
+ if (S_ISDIR(info.st_mode)) {
1005
+ // If entry is a directory, call traverse recursively.
1006
+ if (strcmp(entry->d_name, ".") != 0 && strcmp(entry->d_name, "..") != 0)
1007
+ journal_directory_scan(absolute_path, depth + 1, last_scan_ut);
1008
+
1009
+ }
1010
+ else if (S_ISREG(info.st_mode)) {
1011
+ // If entry is a regular file, check if it ends with .journal.
1012
+ char *filename = entry->d_name;
1013
+ size_t len = strlen(filename);
1014
+
1015
+ if (len > ext_len && strcmp(filename + len - ext_len, ext) == 0) {
1016
+ struct journal_file t = {
1017
+ .file_last_modified_ut = info.st_mtim.tv_sec * USEC_PER_SEC + info.st_mtim.tv_nsec / NSEC_PER_USEC,
1018
+ .last_scan_ut = last_scan_ut,
1019
+ .size = info.st_size,
1020
+ .max_journal_vs_realtime_delta_ut = JOURNAL_VS_REALTIME_DELTA_DEFAULT_UT,
1021
+ };
1022
+ dictionary_set(journal_files_registry, absolute_path, &t, sizeof(t));
1023
+ }
1024
+ }
1025
+ }
1026
+
1027
+ closedir(dir);
1028
+}
1029
+
1030
+static void journal_files_registry_update() {
1031
+ usec_t scan_ut = now_monotonic_usec();
1032
+
1033
+ for(unsigned i = 0; i < MAX_JOURNAL_DIRECTORIES ;i++) {
1034
+ if(!journal_directories[i].path)
1035
+ break;
1036
+
1037
+ journal_directory_scan(journal_directories[i].path, 0, scan_ut);
1038
+ }
1039
+
1040
+ struct journal_file *jf;
1041
+ dfe_start_write(journal_files_registry, jf) {
1042
+ if(jf->last_scan_ut < scan_ut)
1043
+ dictionary_del(journal_files_registry, jf_dfe.name);
1044
+ }
1045
+ dfe_done(jf);
1046
+}
1047
+
1048
+// ----------------------------------------------------------------------------
1049
+
1050
+static bool jf_is_mine(struct journal_file *jf, FUNCTION_QUERY_STATUS *fqs) {
1051
+
1052
+ if((fqs->source_type == SDJF_ALL || (jf->source_type & fqs->source_type) == fqs->source_type) &&
1053
+ (!fqs->source || fqs->source == jf->source)) {
1054
+
1055
+ usec_t anchor_delta = JOURNAL_VS_REALTIME_DELTA_MAX_UT;
1056
+ usec_t first_ut = jf->msg_first_ut;
1057
+ usec_t last_ut = jf->msg_last_ut + anchor_delta;
1058
+
1059
+ if(last_ut >= fqs->after_ut && first_ut <= fqs->before_ut)
1060
+ return true;
1061
+ }
1062
+
1063
+ return false;
1064
+}
1065
+
1066
+static int journal_file_dict_items_backward_compar(const void *a, const void *b) {
1067
+ const DICTIONARY_ITEM **ad = (const DICTIONARY_ITEM **)a, **bd = (const DICTIONARY_ITEM **)b;
1068
+ struct journal_file *jfa = dictionary_acquired_item_value(*ad);
1069
+ struct journal_file *jfb = dictionary_acquired_item_value(*bd);
1070
+
1071
+ if(jfa->msg_last_ut < jfb->msg_last_ut)
1072
+ return 1;
1073
+
1074
+ if(jfa->msg_last_ut > jfb->msg_last_ut)
1075
+ return -1;
1076
+
1077
+ if(jfa->msg_first_ut < jfb->msg_first_ut)
1078
+ return 1;
1079
+
1080
+ if(jfa->msg_first_ut > jfb->msg_first_ut)
1081
+ return -1;
1082
+
1083
+ return 0;
1084
+}
1085
270
- size_t errors_no_timestamp = 0;
271
- size_t row_counter = 0;
272
- size_t rows_added = 0;
1086
+static int journal_file_dict_items_forward_compar(const void *a, const void *b) {
1087
+ return -journal_file_dict_items_backward_compar(a, b);
1088
+}
1089
274
- // the entries are not guaranteed to be sorted, so we process up to 100 entries beyond
275
- // the end of the query to find possibly useful logs for our time-frame
276
- size_t excess_rows_allowed = SYSTEMD_JOURNAL_EXCESS_ROWS_ALLOWED;
1090
+static int netdata_systemd_journal_query(BUFFER *wb, FACETS *facets, FUNCTION_QUERY_STATUS *fqs) {
1091
+ ND_SD_JOURNAL_STATUS status = ND_SD_JOURNAL_NO_FILE_MATCHED;
1092
+ struct journal_file *jf;
1093
278
- ND_SD_JOURNAL_STATUS status = ND_SD_JOURNAL_OK;
1094
+ fqs->files_matched = 0;
1095
+ fqs->file_working = 0;
1096
280
- facets_rows_begin(facets);
281
- while (status == ND_SD_JOURNAL_OK && sd_journal_previous(j) > 0) {
282
- row_counter++;
1097
+ size_t files_used = 0;
1098
+ size_t files_max = dictionary_entries(journal_files_registry);
1099
+ const DICTIONARY_ITEM *file_items[files_max];
1100
284
- usec_t msg_ut;
285
- if(sd_journal_get_realtime_usec(j, &msg_ut) < 0) {
286
- errors_no_timestamp++;
1101
+ // count the files
1102
+ bool files_are_newer = false;
1103
+ dfe_start_read(journal_files_registry, jf) {
1104
+ if(!jf_is_mine(jf, fqs))
1105
continue;
288
- }
1106
290
- if (msg_ut > before_ut || msg_ut >= anchor)
291
- continue;
1107
+ file_items[files_used++] = dictionary_acquired_item_dup(journal_files_registry, jf_dfe.item);
1108
293
- if (msg_ut < after_ut) {
294
- if(--excess_rows_allowed == 0)
295
- break;
1109
+ if(jf->msg_last_ut > fqs->if_modified_since)
1110
+ files_are_newer = true;
1111
+ }
1112
+ dfe_done(jf);
1113
297
- continue;
298
- }
1114
+ fqs->files_matched = files_used;
1115
300
- if(rows_added > entries && --excess_rows_allowed == 0)
301
- break;
1116
+ if(fqs->if_modified_since && !files_are_newer) {
1117
+ buffer_flush(wb);
1118
+ return HTTP_RESP_NOT_MODIFIED;
1119
+ }
1120
303
- netdata_systemd_journal_process_row(j, facets);
304
- facets_row_finished(facets, msg_ut);
305
- rows_added++;
1121
+ // We will not do an if_modified_since query
1122
+ // we know something changed in the files
1123
+ fqs->if_modified_since = 0;
1124
307
- status = check_stop(row_counter, cancelled, stop_monotonic_ut);
1125
+ // sort the files, so that they are optimal for facets
1126
+ if(files_used >= 2) {
1127
+ if (fqs->direction == FACETS_ANCHOR_DIRECTION_BACKWARD)
1128
+ qsort(file_items, files_used, sizeof(const DICTIONARY_ITEM *),
1129
+ journal_file_dict_items_backward_compar);
1130
+ else
1131
+ qsort(file_items, files_used, sizeof(const DICTIONARY_ITEM *),
1132
+ journal_file_dict_items_forward_compar);
1133
}
1134
310
- if(errors_no_timestamp)
311
- netdata_log_error("SYSTEMD-JOURNAL: %zu lines did not have timestamps", errors_no_timestamp);
1135
+ bool partial = false;
1136
+ usec_t started_ut;
1137
+ usec_t ended_ut = now_monotonic_usec();
1138
313
- return status;
314
-}
1139
+ buffer_json_member_add_array(wb, "_journal_files");
1140
+ for(size_t f = 0; f < files_used ;f++) {
1141
+ const char *filename = dictionary_acquired_item_name(file_items[f]);
1142
+ jf = dictionary_acquired_item_value(file_items[f]);
1143
316
-bool netdata_systemd_journal_check_if_modified_since(sd_journal *j, usec_t seek_to, usec_t last_modified) {
317
- // return true, if data have been modified since the timestamp
1144
+ if(!jf_is_mine(jf, fqs))
1145
+ continue;
1146
319
- if(!last_modified || !seek_to)
320
- return false;
1147
+ fqs->file_working++;
1148
+ fqs->cached_count = 0;
1149
322
- if(!netdata_systemd_journal_seek_to(j, seek_to))
323
- return false;
1150
+ size_t rows_useful = fqs->rows_useful;
1151
+ size_t rows_read = fqs->rows_read;
1152
+ size_t bytes_read = fqs->bytes_read;
1153
+ size_t matches_setup_ut = fqs->matches_setup_ut;
1154
325
- usec_t first_msg_ut = 0;
326
- while (sd_journal_previous(j) > 0) {
327
- usec_t msg_ut;
328
- if(sd_journal_get_realtime_usec(j, &msg_ut) < 0)
329
- continue;
1155
+ ND_SD_JOURNAL_STATUS tmp_status = netdata_systemd_journal_query_one_file(filename, wb, facets, jf, fqs);
1156
331
- first_msg_ut = msg_ut;
332
- break;
333
- }
1157
+ rows_useful = fqs->rows_useful - rows_useful;
1158
+ rows_read = fqs->rows_read - rows_read;
1159
+ bytes_read = fqs->bytes_read - bytes_read;
1160
+ matches_setup_ut = fqs->matches_setup_ut - matches_setup_ut;
1161
335
- return first_msg_ut != last_modified;
336
-}
1162
+ started_ut = ended_ut;
1163
+ ended_ut = now_monotonic_usec();
1164
+ usec_t duration_ut = ended_ut - started_ut;
1165
338
-static int netdata_systemd_journal_query(BUFFER *wb, FACETS *facets,
339
- usec_t after_ut, usec_t before_ut,
340
- usec_t anchor, FACETS_ANCHOR_DIRECTION direction, size_t entries,
341
- usec_t if_modified_since, bool data_only,
342
- usec_t stop_monotonic_ut,
343
- bool *cancelled) {
344
- sd_journal *j = netdata_open_systemd_journal();
345
- if(!j)
346
- return HTTP_RESP_INTERNAL_SERVER_ERROR;
1166
+ buffer_json_add_array_item_object(wb); // journal file
1167
+ {
1168
+ // information about the file
1169
+ buffer_json_member_add_string(wb, "_filename", filename);
1170
+ buffer_json_member_add_uint64(wb, "_source_type", jf->source_type);
1171
+ buffer_json_member_add_string(wb, "_source", string2str(jf->source));
1172
+ buffer_json_member_add_uint64(wb, "_last_modified_ut", jf->file_last_modified_ut);
1173
+ buffer_json_member_add_uint64(wb, "_msg_first_ut", jf->msg_first_ut);
1174
+ buffer_json_member_add_uint64(wb, "_msg_last_ut", jf->msg_last_ut);
1175
+ buffer_json_member_add_uint64(wb, "_journal_vs_realtime_delta_ut", jf->max_journal_vs_realtime_delta_ut);
1176
+
1177
+ // information about the current use of the file
1178
+ buffer_json_member_add_uint64(wb, "duration_ut", ended_ut - started_ut);
1179
+ buffer_json_member_add_uint64(wb, "rows_read", rows_read);
1180
+ buffer_json_member_add_uint64(wb, "rows_useful", rows_useful);
1181
+ buffer_json_member_add_double(wb, "rows_per_second", (double) rows_read / (double) duration_ut * (double) USEC_PER_SEC);
1182
+ buffer_json_member_add_uint64(wb, "bytes_read", bytes_read);
1183
+ buffer_json_member_add_double(wb, "bytes_per_second", (double) bytes_read / (double) duration_ut * (double) USEC_PER_SEC);
1184
+ buffer_json_member_add_uint64(wb, "duration_matches_ut", matches_setup_ut);
1185
+ }
1186
+ buffer_json_object_close(wb); // journal file
1187
348
- usec_t last_modified = 0;
1188
+ bool stop = false;
1189
+ switch(tmp_status) {
1190
+ case ND_SD_JOURNAL_OK:
1191
+ case ND_SD_JOURNAL_NO_FILE_MATCHED:
1192
+ status = (status == ND_SD_JOURNAL_OK) ? ND_SD_JOURNAL_OK : tmp_status;
1193
+ break;
1194
350
- ND_SD_JOURNAL_STATUS status;
1195
+ case ND_SD_JOURNAL_FAILED_TO_OPEN:
1196
+ case ND_SD_JOURNAL_FAILED_TO_SEEK:
1197
+ partial = true;
1198
+ if(status == ND_SD_JOURNAL_NO_FILE_MATCHED)
1199
+ status = tmp_status;
1200
+ break;
1201
352
- if(data_only && anchor /* && !netdata_systemd_journal_check_if_modified_since(j, before_ut, if_modified_since) */) {
353
- facets_data_only_mode(facets);
1202
+ case ND_SD_JOURNAL_CANCELLED:
1203
+ case ND_SD_JOURNAL_TIMED_OUT:
1204
+ partial = true;
1205
+ stop = true;
1206
+ status = tmp_status;
1207
+ break;
1208
355
- // we can do a data-only query
356
- if(direction == FACETS_ANCHOR_DIRECTION_FORWARD)
357
- status = netdata_systemd_journal_query_data_forward(j, wb, facets, after_ut, before_ut, anchor, entries, stop_monotonic_ut, cancelled);
358
- else
359
- status = netdata_systemd_journal_query_data_backward(j, wb, facets, after_ut, before_ut, anchor, entries, stop_monotonic_ut, cancelled);
360
- }
361
- else {
362
- // we have to do a full query
363
- status = netdata_systemd_journal_query_full(j, wb, facets,
364
- after_ut, before_ut, if_modified_since,
365
- stop_monotonic_ut, &last_modified, cancelled);
1209
+ case ND_SD_JOURNAL_NOT_MODIFIED:
1210
+ internal_fatal(true, "this should never be returned here");
1211
+ break;
1212
+ }
1213
+
1214
+ if(stop)
1215
+ break;
1216
}
1217
+ buffer_json_array_close(wb); // _journal_files
1218
368
- sd_journal_close(j);
1219
+ // release the files
1220
+ for(size_t f = 0; f < files_used ;f++)
1221
+ dictionary_acquired_item_release(journal_files_registry, file_items[f]);
1222
370
- if(status != ND_SD_JOURNAL_OK && status != ND_SD_JOURNAL_TIMED_OUT) {
371
- buffer_flush(wb);
1223
+ switch (status) {
1224
+ case ND_SD_JOURNAL_OK:
1225
+ case ND_SD_JOURNAL_TIMED_OUT:
1226
+ case ND_SD_JOURNAL_NO_FILE_MATCHED:
1227
+ break;
1228
373
- switch (status) {
374
- case ND_SD_JOURNAL_CANCELLED:
375
- return HTTP_RESP_CLIENT_CLOSED_REQUEST;
1229
+ case ND_SD_JOURNAL_CANCELLED:
1230
+ buffer_flush(wb);
1231
+ return HTTP_RESP_CLIENT_CLOSED_REQUEST;
1232
377
- case ND_SD_JOURNAL_NOT_MODIFIED:
378
- return HTTP_RESP_NOT_MODIFIED;
1233
+ case ND_SD_JOURNAL_NOT_MODIFIED:
1234
+ buffer_flush(wb);
1235
+ return HTTP_RESP_NOT_MODIFIED;
1236
380
- default:
381
- case ND_SD_JOURNAL_FAILED_TO_SEEK:
382
- return HTTP_RESP_INTERNAL_SERVER_ERROR;
383
- }
1237
+ default:
1238
+ case ND_SD_JOURNAL_FAILED_TO_OPEN:
1239
+ case ND_SD_JOURNAL_FAILED_TO_SEEK:
1240
+ buffer_flush(wb);
1241
+ return HTTP_RESP_INTERNAL_SERVER_ERROR;
1242
}
1243
1244
buffer_json_member_add_uint64(wb, "status", HTTP_RESP_OK);
387
- buffer_json_member_add_boolean(wb, "partial", status != ND_SD_JOURNAL_OK);
1245
+ buffer_json_member_add_boolean(wb, "partial", partial);
1246
buffer_json_member_add_string(wb, "type", "table");
1247
390
- if(!data_only) {
1248
+ if(!fqs->data_only) {
1249
buffer_json_member_add_time_t(wb, "update_every", 1);
1250
buffer_json_member_add_string(wb, "help", SYSTEMD_JOURNAL_FUNCTION_DESCRIPTION);
393
- buffer_json_member_add_uint64(wb, "last_modified", last_modified);
1251
}
1252
396
- facets_report(facets, wb);
1253
+ if(!fqs->data_only || fqs->tail)
1254
+ buffer_json_member_add_uint64(wb, "last_modified", fqs->last_modified);
1255
+
1256
+ facets_sort_and_reorder_keys(facets);
1257
+ facets_report(facets, wb, used_hashes_registry);
1258
398
- buffer_json_member_add_time_t(wb, "expires", now_realtime_sec() + (data_only ? 3600 : 0));
1259
+ buffer_json_member_add_time_t(wb, "expires", now_realtime_sec() + (fqs->data_only ? 3600 : 0));
1260
buffer_json_finalize(wb);
1261
1262
return HTTP_RESP_OK;
1269
"\n"
1270
"%s\n"
1271
"\n"
411
- "The following filters are supported:\n"
1272
+ "The following parameters are supported:\n"
1273
"\n"
413
- " help\n"
1274
+ " "JOURNAL_PARAMETER_HELP"\n"
1275
" Shows this help message.\n"
1276
"\n"
416
- " before:TIMESTAMP\n"
1277
+ " "JOURNAL_PARAMETER_ID":STRING\n"
1278
+ " Caller supplied unique ID of the request.\n"
1279
+ " This can be used later to request a progress report of the query.\n"
1280
+ " Optional, but if omitted no `"JOURNAL_PARAMETER_PROGRESS"` can be requested.\n"
1281
+ "\n"
1282
+ " "JOURNAL_PARAMETER_INFO"\n"
1283
+ " Request initial configuration information about the plugin.\n"
1284
+ " The key entity returned is the required_params array, which includes\n"
1285
+ " all the available systemd journal sources.\n"
1286
+ " When `"JOURNAL_PARAMETER_INFO"` is requested, all other parameters are ignored.\n"
1287
+ "\n"
1288
+ " "JOURNAL_PARAMETER_DELTA"\n"
1289
+ " When doing data queries, include deltas for histogram and facets.\n"
1290
+ "\n"
1291
+ " "JOURNAL_PARAMETER_TAIL"\n"
1292
+ " Do a tail query, to return the newest items between the anchor and before.\n"
1293
+ "\n"
1294
+ " "JOURNAL_PARAMETER_PROGRESS"\n"
1295
+ " Request a progress report (the `id` of a running query is required).\n"
1296
+ " When `"JOURNAL_PARAMETER_PROGRESS"` is requested, only parameter `"JOURNAL_PARAMETER_ID"` is used.\n"
1297
+ "\n"
1298
+ " "JOURNAL_PARAMETER_DATA_ONLY"\n"
1299
+ " Quickly respond with data requested, without generating a\n"
1300
+ " histogram and facets counters.\n"
1301
+ "\n"
1302
+ " "JOURNAL_PARAMETER_SLICE":true or "JOURNAL_PARAMETER_SLICE":false\n"
1303
+ " When it is turned on, the plugin is executing filtering via libsystemd,\n"
1304
+ " utilizing all the available indexes of the journal files.\n"
1305
+ " When it is off, only the time constraint is handled by libsystemd and\n"
1306
+ " all filtering is done by the plugin.\n"
1307
+ " The default is: %s\n"
1308
+ "\n"
1309
+ " "JOURNAL_PARAMETER_SOURCE":SOURCE\n"
1310
+ " Query only the specified journal sources.\n"
1311
+ " Do an `"JOURNAL_PARAMETER_INFO"` query to find the sources.\n"
1312
+ "\n"
1313
+ " "JOURNAL_PARAMETER_BEFORE":TIMESTAMP_IN_SECONDS\n"
1314
" Absolute or relative (to now) timestamp in seconds, to start the query.\n"
1315
" The query is always executed from the most recent to the oldest log entry.\n"
1316
" If not given the default is: now.\n"
1317
"\n"
421
- " after:TIMESTAMP\n"
1318
+ " "JOURNAL_PARAMETER_AFTER":TIMESTAMP_IN_SECONDS\n"
1319
" Absolute or relative (to `before`) timestamp in seconds, to end the query.\n"
1320
" If not given, the default is %d.\n"
1321
"\n"
425
- " last:ITEMS\n"
1322
+ " "JOURNAL_PARAMETER_LAST":ITEMS\n"
1323
" The number of items to return.\n"
1324
" The default is %d.\n"
1325
"\n"
429
- " anchor:NUMBER\n"
430
- " The `timestamp` of the item last received, to return log entries after that.\n"
431
- " If not given, the query will return the top `ITEMS` from the most recent.\n"
1326
+ " "JOURNAL_PARAMETER_ANCHOR":TIMESTAMP_IN_MICROSECONDS\n"
1327
+ " Return items relative to this timestamp.\n"
1328
+ " The exact items to be returned depend on the query `"JOURNAL_PARAMETER_DIRECTION"`.\n"
1329
+ "\n"
1330
+ " "JOURNAL_PARAMETER_DIRECTION":forward or "JOURNAL_PARAMETER_DIRECTION":backward\n"
1331
+ " When set to `backward` (default) the items returned are the newest before the\n"
1332
+ " `"JOURNAL_PARAMETER_ANCHOR"`, (or `"JOURNAL_PARAMETER_BEFORE"` if `"JOURNAL_PARAMETER_ANCHOR"` is not set)\n"
1333
+ " When set to `forward` the items returned are the oldest after the\n"
1334
+ " `"JOURNAL_PARAMETER_ANCHOR"`, (or `"JOURNAL_PARAMETER_AFTER"` if `"JOURNAL_PARAMETER_ANCHOR"` is not set)\n"
1335
+ " The default is: %s\n"
1336
+ "\n"
1337
+ " "JOURNAL_PARAMETER_QUERY":SIMPLE_PATTERN\n"
1338
+ " Do a full text search to find the log entries matching the pattern given.\n"
1339
+ " The plugin is searching for matches on all fields of the database.\n"
1340
+ "\n"
1341
+ " "JOURNAL_PARAMETER_IF_MODIFIED_SINCE":TIMESTAMP_IN_MICROSECONDS\n"
1342
+ " Each successful response, includes a `last_modified` field.\n"
1343
+ " By providing the timestamp to the `"JOURNAL_PARAMETER_IF_MODIFIED_SINCE"` parameter,\n"
1344
+ " the plugin will return 200 with a successful response, or 304 if the source has not\n"
1345
+ " been modified since that timestamp.\n"
1346
+ "\n"
1347
+ " "JOURNAL_PARAMETER_HISTOGRAM":facet_id\n"
1348
+ " Use the given `facet_id` for the histogram.\n"
1349
+ " This parameter is ignored in `"JOURNAL_PARAMETER_DATA_ONLY"` mode.\n"
1350
+ "\n"
1351
+ " "JOURNAL_PARAMETER_FACETS":facet_id1,facet_id2,facet_id3,...\n"
1352
+ " Add the given facets to the list of fields for which analysis is required.\n"
1353
+ " The plugin will offer both a histogram and facet value counters for its values.\n"
1354
+ " This parameter is ignored in `"JOURNAL_PARAMETER_DATA_ONLY"` mode.\n"
1355
"\n"
1356
" facet_id:value_id1,value_id2,value_id3,...\n"
1357
" Apply filters to the query, based on the facet IDs returned.\n"
1358
" Each `facet_id` can be given once, but multiple `facet_ids` can be given.\n"
1359
"\n"
437
- "Filters can be combined. Each filter can be given only one time.\n"
1360
+ " There is special mode. By specifying:\n"
1361
+ "\n"
1362
+ " - `"JOURNAL_PARAMETER_DIRECTION":forward`,\n"
1363
+ " - `"JOURNAL_PARAMETER_ANCHOR":TIMESTAMP_IN_USEC`,\n"
1364
+ " - `"JOURNAL_PARAMETER_DATA_ONLY"`, and\n"
1365
+ " - `"JOURNAL_PARAMETER_IF_MODIFIED_SINCE":TIMESTAMP_IN_USEC`\n"
1366
+ "\n"
1367
, program_name
1368
, SYSTEMD_JOURNAL_FUNCTION_NAME
1369
, SYSTEMD_JOURNAL_FUNCTION_DESCRIPTION
1370
+ , JOURNAL_DEFAULT_SLICE_MODE ? "true" : "false" // slice
1371
, -SYSTEMD_JOURNAL_DEFAULT_QUERY_DURATION
1372
, SYSTEMD_JOURNAL_DEFAULT_ITEMS_PER_QUERY
1373
+ , JOURNAL_DEFAULT_DIRECTION == FACETS_ANCHOR_DIRECTION_BACKWARD ? "backward" : "forward"
1374
);
1375
1376
netdata_mutex_lock(&stdout_mutex);
1380
buffer_free(wb);
1381
}
1382
1383
+const char *errno_map[] = {
1384
+ [1] = "1 (EPERM)", // "Operation not permitted",
1385
+ [2] = "2 (ENOENT)", // "No such file or directory",
1386
+ [3] = "3 (ESRCH)", // "No such process",
1387
+ [4] = "4 (EINTR)", // "Interrupted system call",
1388
+ [5] = "5 (EIO)", // "Input/output error",
1389
+ [6] = "6 (ENXIO)", // "No such device or address",
1390
+ [7] = "7 (E2BIG)", // "Argument list too long",
1391
+ [8] = "8 (ENOEXEC)", // "Exec format error",
1392
+ [9] = "9 (EBADF)", // "Bad file descriptor",
1393
+ [10] = "10 (ECHILD)", // "No child processes",
1394
+ [11] = "11 (EAGAIN)", // "Resource temporarily unavailable",
1395
+ [12] = "12 (ENOMEM)", // "Cannot allocate memory",
1396
+ [13] = "13 (EACCES)", // "Permission denied",
1397
+ [14] = "14 (EFAULT)", // "Bad address",
1398
+ [15] = "15 (ENOTBLK)", // "Block device required",
1399
+ [16] = "16 (EBUSY)", // "Device or resource busy",
1400
+ [17] = "17 (EEXIST)", // "File exists",
1401
+ [18] = "18 (EXDEV)", // "Invalid cross-device link",
1402
+ [19] = "19 (ENODEV)", // "No such device",
1403
+ [20] = "20 (ENOTDIR)", // "Not a directory",
1404
+ [21] = "21 (EISDIR)", // "Is a directory",
1405
+ [22] = "22 (EINVAL)", // "Invalid argument",
1406
+ [23] = "23 (ENFILE)", // "Too many open files in system",
1407
+ [24] = "24 (EMFILE)", // "Too many open files",
1408
+ [25] = "25 (ENOTTY)", // "Inappropriate ioctl for device",
1409
+ [26] = "26 (ETXTBSY)", // "Text file busy",
1410
+ [27] = "27 (EFBIG)", // "File too large",
1411
+ [28] = "28 (ENOSPC)", // "No space left on device",
1412
+ [29] = "29 (ESPIPE)", // "Illegal seek",
1413
+ [30] = "30 (EROFS)", // "Read-only file system",
1414
+ [31] = "31 (EMLINK)", // "Too many links",
1415
+ [32] = "32 (EPIPE)", // "Broken pipe",
1416
+ [33] = "33 (EDOM)", // "Numerical argument out of domain",
1417
+ [34] = "34 (ERANGE)", // "Numerical result out of range",
1418
+ [35] = "35 (EDEADLK)", // "Resource deadlock avoided",
1419
+ [36] = "36 (ENAMETOOLONG)", // "File name too long",
1420
+ [37] = "37 (ENOLCK)", // "No locks available",
1421
+ [38] = "38 (ENOSYS)", // "Function not implemented",
1422
+ [39] = "39 (ENOTEMPTY)", // "Directory not empty",
1423
+ [40] = "40 (ELOOP)", // "Too many levels of symbolic links",
1424
+ [42] = "42 (ENOMSG)", // "No message of desired type",
1425
+ [43] = "43 (EIDRM)", // "Identifier removed",
1426
+ [44] = "44 (ECHRNG)", // "Channel number out of range",
1427
+ [45] = "45 (EL2NSYNC)", // "Level 2 not synchronized",
1428
+ [46] = "46 (EL3HLT)", // "Level 3 halted",
1429
+ [47] = "47 (EL3RST)", // "Level 3 reset",
1430
+ [48] = "48 (ELNRNG)", // "Link number out of range",
1431
+ [49] = "49 (EUNATCH)", // "Protocol driver not attached",
1432
+ [50] = "50 (ENOCSI)", // "No CSI structure available",
1433
+ [51] = "51 (EL2HLT)", // "Level 2 halted",
1434
+ [52] = "52 (EBADE)", // "Invalid exchange",
1435
+ [53] = "53 (EBADR)", // "Invalid request descriptor",
1436
+ [54] = "54 (EXFULL)", // "Exchange full",
1437
+ [55] = "55 (ENOANO)", // "No anode",
1438
+ [56] = "56 (EBADRQC)", // "Invalid request code",
1439
+ [57] = "57 (EBADSLT)", // "Invalid slot",
1440
+ [59] = "59 (EBFONT)", // "Bad font file format",
1441
+ [60] = "60 (ENOSTR)", // "Device not a stream",
1442
+ [61] = "61 (ENODATA)", // "No data available",
1443
+ [62] = "62 (ETIME)", // "Timer expired",
1444
+ [63] = "63 (ENOSR)", // "Out of streams resources",
1445
+ [64] = "64 (ENONET)", // "Machine is not on the network",
1446
+ [65] = "65 (ENOPKG)", // "Package not installed",
1447
+ [66] = "66 (EREMOTE)", // "Object is remote",
1448
+ [67] = "67 (ENOLINK)", // "Link has been severed",
1449
+ [68] = "68 (EADV)", // "Advertise error",
1450
+ [69] = "69 (ESRMNT)", // "Srmount error",
1451
+ [70] = "70 (ECOMM)", // "Communication error on send",
1452
+ [71] = "71 (EPROTO)", // "Protocol error",
1453
+ [72] = "72 (EMULTIHOP)", // "Multihop attempted",
1454
+ [73] = "73 (EDOTDOT)", // "RFS specific error",
1455
+ [74] = "74 (EBADMSG)", // "Bad message",
1456
+ [75] = "75 (EOVERFLOW)", // "Value too large for defined data type",
1457
+ [76] = "76 (ENOTUNIQ)", // "Name not unique on network",
1458
+ [77] = "77 (EBADFD)", // "File descriptor in bad state",
1459
+ [78] = "78 (EREMCHG)", // "Remote address changed",
1460
+ [79] = "79 (ELIBACC)", // "Can not access a needed shared library",
1461
+ [80] = "80 (ELIBBAD)", // "Accessing a corrupted shared library",
1462
+ [81] = "81 (ELIBSCN)", // ".lib section in a.out corrupted",
1463
+ [82] = "82 (ELIBMAX)", // "Attempting to link in too many shared libraries",
1464
+ [83] = "83 (ELIBEXEC)", // "Cannot exec a shared library directly",
1465
+ [84] = "84 (EILSEQ)", // "Invalid or incomplete multibyte or wide character",
1466
+ [85] = "85 (ERESTART)", // "Interrupted system call should be restarted",
1467
+ [86] = "86 (ESTRPIPE)", // "Streams pipe error",
1468
+ [87] = "87 (EUSERS)", // "Too many users",
1469
+ [88] = "88 (ENOTSOCK)", // "Socket operation on non-socket",
1470
+ [89] = "89 (EDESTADDRREQ)", // "Destination address required",
1471
+ [90] = "90 (EMSGSIZE)", // "Message too long",
1472
+ [91] = "91 (EPROTOTYPE)", // "Protocol wrong type for socket",
1473
+ [92] = "92 (ENOPROTOOPT)", // "Protocol not available",
1474
+ [93] = "93 (EPROTONOSUPPORT)", // "Protocol not supported",
1475
+ [94] = "94 (ESOCKTNOSUPPORT)", // "Socket type not supported",
1476
+ [95] = "95 (ENOTSUP)", // "Operation not supported",
1477
+ [96] = "96 (EPFNOSUPPORT)", // "Protocol family not supported",
1478
+ [97] = "97 (EAFNOSUPPORT)", // "Address family not supported by protocol",
1479
+ [98] = "98 (EADDRINUSE)", // "Address already in use",
1480
+ [99] = "99 (EADDRNOTAVAIL)", // "Cannot assign requested address",
1481
+ [100] = "100 (ENETDOWN)", // "Network is down",
1482
+ [101] = "101 (ENETUNREACH)", // "Network is unreachable",
1483
+ [102] = "102 (ENETRESET)", // "Network dropped connection on reset",
1484
+ [103] = "103 (ECONNABORTED)", // "Software caused connection abort",
1485
+ [104] = "104 (ECONNRESET)", // "Connection reset by peer",
1486
+ [105] = "105 (ENOBUFS)", // "No buffer space available",
1487
+ [106] = "106 (EISCONN)", // "Transport endpoint is already connected",
1488
+ [107] = "107 (ENOTCONN)", // "Transport endpoint is not connected",
1489
+ [108] = "108 (ESHUTDOWN)", // "Cannot send after transport endpoint shutdown",
1490
+ [109] = "109 (ETOOMANYREFS)", // "Too many references: cannot splice",
1491
+ [110] = "110 (ETIMEDOUT)", // "Connection timed out",
1492
+ [111] = "111 (ECONNREFUSED)", // "Connection refused",
1493
+ [112] = "112 (EHOSTDOWN)", // "Host is down",
1494
+ [113] = "113 (EHOSTUNREACH)", // "No route to host",
1495
+ [114] = "114 (EALREADY)", // "Operation already in progress",
1496
+ [115] = "115 (EINPROGRESS)", // "Operation now in progress",
1497
+ [116] = "116 (ESTALE)", // "Stale file handle",
1498
+ [117] = "117 (EUCLEAN)", // "Structure needs cleaning",
1499
+ [118] = "118 (ENOTNAM)", // "Not a XENIX named type file",
1500
+ [119] = "119 (ENAVAIL)", // "No XENIX semaphores available",
1501
+ [120] = "120 (EISNAM)", // "Is a named type file",
1502
+ [121] = "121 (EREMOTEIO)", // "Remote I/O error",
1503
+ [122] = "122 (EDQUOT)", // "Disk quota exceeded",
1504
+ [123] = "123 (ENOMEDIUM)", // "No medium found",
1505
+ [124] = "124 (EMEDIUMTYPE)", // "Wrong medium type",
1506
+ [125] = "125 (ECANCELED)", // "Operation canceled",
1507
+ [126] = "126 (ENOKEY)", // "Required key not available",
1508
+ [127] = "127 (EKEYEXPIRED)", // "Key has expired",
1509
+ [128] = "128 (EKEYREVOKED)", // "Key has been revoked",
1510
+ [129] = "129 (EKEYREJECTED)", // "Key was rejected by service",
1511
+ [130] = "130 (EOWNERDEAD)", // "Owner died",
1512
+ [131] = "131 (ENOTRECOVERABLE)", // "State not recoverable",
1513
+ [132] = "132 (ERFKILL)", // "Operation not possible due to RF-kill",
1514
+ [133] = "133 (EHWPOISON)", // "Memory page has hardware error",
1515
+};
1516
+
1517
static const char *syslog_facility_to_name(int facility) {
1518
switch (facility) {
1519
case LOG_FAC(LOG_KERN): return "kern";
1554
}
1555
}
1556
492
-static FACET_ROW_SEVERITY syslog_priority_to_facet_severity(int priority) {
1557
+static FACET_ROW_SEVERITY syslog_priority_to_facet_severity(FACETS *facets __maybe_unused, FACET_ROW *row, void *data __maybe_unused) {
1558
// same to
1559
// https://github.com/systemd/systemd/blob/aab9e4b2b86905a15944a1ac81e471b5b7075932/src/basic/terminal-util.c#L1501
1560
// function get_log_colors()
1561
1562
+ FACET_ROW_KEY_VALUE *priority_rkv = dictionary_get(row->dict, "PRIORITY");
1563
+ if(!priority_rkv || priority_rkv->empty)
1564
+ return FACET_ROW_SEVERITY_NORMAL;
1565
+
1566
+ int priority = str2i(buffer_tostring(priority_rkv->wb));
1567
+
1568
if(priority <= LOG_ERR)
1569
return FACET_ROW_SEVERITY_CRITICAL;
1570
1604
return buffer;
1605
}
1606
536
-static void netdata_systemd_journal_transform_syslog_facility(FACETS *facets __maybe_unused, BUFFER *wb, void *data __maybe_unused) {
1607
+static void netdata_systemd_journal_transform_syslog_facility(FACETS *facets __maybe_unused, BUFFER *wb, FACETS_TRANSFORMATION_SCOPE scope __maybe_unused, void *data __maybe_unused) {
1608
const char *v = buffer_tostring(wb);
1609
if(*v && isdigit(*v)) {
1610
int facility = str2i(buffer_tostring(wb));
1616
}
1617
}
1618
548
-static void netdata_systemd_journal_transform_priority(FACETS *facets __maybe_unused, BUFFER *wb, void *data __maybe_unused) {
1619
+static void netdata_systemd_journal_transform_priority(FACETS *facets __maybe_unused, BUFFER *wb, FACETS_TRANSFORMATION_SCOPE scope __maybe_unused, void *data __maybe_unused) {
1620
+ if(scope == FACETS_TRANSFORM_FACET_SORT)
1621
+ return;
1622
+
1623
const char *v = buffer_tostring(wb);
1624
if(*v && isdigit(*v)) {
1625
int priority = str2i(buffer_tostring(wb));
1628
buffer_flush(wb);
1629
buffer_strcat(wb, name);
1630
}
1631
+ }
1632
+}
1633
558
- facets_set_current_row_severity(facets, syslog_priority_to_facet_severity(priority));
1634
+static void netdata_systemd_journal_transform_errno(FACETS *facets __maybe_unused, BUFFER *wb, FACETS_TRANSFORMATION_SCOPE scope __maybe_unused, void *data __maybe_unused) {
1635
+ if(scope == FACETS_TRANSFORM_FACET_SORT)
1636
+ return;
1637
+
1638
+ const char *v = buffer_tostring(wb);
1639
+ if(*v && isdigit(*v)) {
1640
+ unsigned err_no = str2u(buffer_tostring(wb));
1641
+ if(err_no > 0 && err_no < sizeof(errno_map) / sizeof(*errno_map)) {
1642
+ const char *name = errno_map[err_no];
1643
+ if(name) {
1644
+ buffer_flush(wb);
1645
+ buffer_strcat(wb, name);
1646
+ }
1647
+ }
1648
}
1649
}
1650
1726
return (*e)->str;
1727
}
1728
640
-static void netdata_systemd_journal_transform_uid(FACETS *facets __maybe_unused, BUFFER *wb, void *data __maybe_unused) {
1729
+DICTIONARY *boot_ids_to_first_ut = NULL;
1730
+
1731
+static void netdata_systemd_journal_transform_boot_id(FACETS *facets __maybe_unused, BUFFER *wb, FACETS_TRANSFORMATION_SCOPE scope __maybe_unused, void *data __maybe_unused) {
1732
+ const char *boot_id = buffer_tostring(wb);
1733
+ if(*boot_id && isxdigit(*boot_id)) {
1734
+ usec_t ut = UINT64_MAX;
1735
+ usec_t *p_ut = dictionary_get(boot_ids_to_first_ut, boot_id);
1736
+ if(!p_ut) {
1737
+ struct journal_file *jf;
1738
+ dfe_start_read(journal_files_registry, jf) {
1739
+ const char *files[2] = {
1740
+ [0] = jf_dfe.name,
1741
+ [1] = NULL,
1742
+ };
1743
+
1744
+ sd_journal *j = NULL;
1745
+ if(sd_journal_open_files(&j, files, ND_SD_JOURNAL_OPEN_FLAGS) < 0 || !j)
1746
+ continue;
1747
+
1748
+ char m[100];
1749
+ size_t len = snprintfz(m, sizeof(m), "_BOOT_ID=%s", boot_id);
1750
+ usec_t t_ut = 0;
1751
+ if(sd_journal_add_match(j, m, len) < 0 ||
1752
+ sd_journal_seek_head(j) < 0 ||
1753
+ sd_journal_next(j) < 0 ||
1754
+ sd_journal_get_realtime_usec(j, &t_ut) < 0 || !t_ut) {
1755
+ sd_journal_close(j);
1756
+ continue;
1757
+ }
1758
+
1759
+ if(t_ut < ut)
1760
+ ut = t_ut;
1761
+
1762
+ sd_journal_close(j);
1763
+ }
1764
+ dfe_done(jf);
1765
+
1766
+ dictionary_set(boot_ids_to_first_ut, boot_id, &ut, sizeof(ut));
1767
+ }
1768
+ else
1769
+ ut = *p_ut;
1770
+
1771
+ if(ut != UINT64_MAX) {
1772
+ time_t timestamp_sec = (time_t)(ut / USEC_PER_SEC);
1773
+ struct tm tm;
1774
+ char buffer[30];
1775
+
1776
+ gmtime_r(×tamp_sec, &tm);
1777
+ strftime(buffer, sizeof(buffer), "%Y-%m-%d %H:%M:%S", &tm);
1778
+
1779
+ switch(scope) {
1780
+ default:
1781
+ case FACETS_TRANSFORM_DATA:
1782
+ case FACETS_TRANSFORM_VALUE:
1783
+ buffer_sprintf(wb, " (%s UTC) ", buffer);
1784
+ break;
1785
+
1786
+ case FACETS_TRANSFORM_FACET:
1787
+ case FACETS_TRANSFORM_FACET_SORT:
1788
+ case FACETS_TRANSFORM_HISTOGRAM:
1789
+ buffer_flush(wb);
1790
+ buffer_sprintf(wb, "%s UTC", buffer);
1791
+ break;
1792
+ }
1793
+ }
1794
+ }
1795
+}
1796
+
1797
+static void netdata_systemd_journal_transform_uid(FACETS *facets __maybe_unused, BUFFER *wb, FACETS_TRANSFORMATION_SCOPE scope __maybe_unused, void *data __maybe_unused) {
1798
+ if(scope == FACETS_TRANSFORM_FACET_SORT)
1799
+ return;
1800
+
1801
const char *v = buffer_tostring(wb);
1802
if(*v && isdigit(*v)) {
1803
uid_t uid = str2i(buffer_tostring(wb));
1807
}
1808
}
1809
650
-static void netdata_systemd_journal_transform_gid(FACETS *facets __maybe_unused, BUFFER *wb, void *data __maybe_unused) {
1810
+static void netdata_systemd_journal_transform_gid(FACETS *facets __maybe_unused, BUFFER *wb, FACETS_TRANSFORMATION_SCOPE scope __maybe_unused, void *data __maybe_unused) {
1811
+ if(scope == FACETS_TRANSFORM_FACET_SORT)
1812
+ return;
1813
+
1814
const char *v = buffer_tostring(wb);
1815
if(*v && isdigit(*v)) {
1816
gid_t gid = str2i(buffer_tostring(wb));
1820
}
1821
}
1822
1823
+const char *linux_capabilities[] = {
1824
+ [CAP_CHOWN] = "CHOWN",
1825
+ [CAP_DAC_OVERRIDE] = "DAC_OVERRIDE",
1826
+ [CAP_DAC_READ_SEARCH] = "DAC_READ_SEARCH",
1827
+ [CAP_FOWNER] = "FOWNER",
1828
+ [CAP_FSETID] = "FSETID",
1829
+ [CAP_KILL] = "KILL",
1830
+ [CAP_SETGID] = "SETGID",
1831
+ [CAP_SETUID] = "SETUID",
1832
+ [CAP_SETPCAP] = "SETPCAP",
1833
+ [CAP_LINUX_IMMUTABLE] = "LINUX_IMMUTABLE",
1834
+ [CAP_NET_BIND_SERVICE] = "NET_BIND_SERVICE",
1835
+ [CAP_NET_BROADCAST] = "NET_BROADCAST",
1836
+ [CAP_NET_ADMIN] = "NET_ADMIN",
1837
+ [CAP_NET_RAW] = "NET_RAW",
1838
+ [CAP_IPC_LOCK] = "IPC_LOCK",
1839
+ [CAP_IPC_OWNER] = "IPC_OWNER",
1840
+ [CAP_SYS_MODULE] = "SYS_MODULE",
1841
+ [CAP_SYS_RAWIO] = "SYS_RAWIO",
1842
+ [CAP_SYS_CHROOT] = "SYS_CHROOT",
1843
+ [CAP_SYS_PTRACE] = "SYS_PTRACE",
1844
+ [CAP_SYS_PACCT] = "SYS_PACCT",
1845
+ [CAP_SYS_ADMIN] = "SYS_ADMIN",
1846
+ [CAP_SYS_BOOT] = "SYS_BOOT",
1847
+ [CAP_SYS_NICE] = "SYS_NICE",
1848
+ [CAP_SYS_RESOURCE] = "SYS_RESOURCE",
1849
+ [CAP_SYS_TIME] = "SYS_TIME",
1850
+ [CAP_SYS_TTY_CONFIG] = "SYS_TTY_CONFIG",
1851
+ [CAP_MKNOD] = "MKNOD",
1852
+ [CAP_LEASE] = "LEASE",
1853
+ [CAP_AUDIT_WRITE] = "AUDIT_WRITE",
1854
+ [CAP_AUDIT_CONTROL] = "AUDIT_CONTROL",
1855
+ [CAP_SETFCAP] = "SETFCAP",
1856
+ [CAP_MAC_OVERRIDE] = "MAC_OVERRIDE",
1857
+ [CAP_MAC_ADMIN] = "MAC_ADMIN",
1858
+ [CAP_SYSLOG] = "SYSLOG",
1859
+ [CAP_WAKE_ALARM] = "WAKE_ALARM",
1860
+ [CAP_BLOCK_SUSPEND] = "BLOCK_SUSPEND",
1861
+ [37 /*CAP_AUDIT_READ*/] = "AUDIT_READ",
1862
+ [38 /*CAP_PERFMON*/] = "PERFMON",
1863
+ [39 /*CAP_BPF*/] = "BPF",
1864
+ [40 /* CAP_CHECKPOINT_RESTORE */] = "CHECKPOINT_RESTORE",
1865
+};
1866
+
1867
+static void netdata_systemd_journal_transform_cap_effective(FACETS *facets __maybe_unused, BUFFER *wb, FACETS_TRANSFORMATION_SCOPE scope __maybe_unused, void *data __maybe_unused) {
1868
+ if(scope == FACETS_TRANSFORM_FACET_SORT)
1869
+ return;
1870
+
1871
+ const char *v = buffer_tostring(wb);
1872
+ if(*v && isdigit(*v)) {
1873
+ uint64_t cap = strtoul(buffer_tostring(wb), NULL, 16);
1874
+ if(cap) {
1875
+ buffer_fast_strcat(wb, " (", 2);
1876
+ for (size_t i = 0, added = 0; i < sizeof(linux_capabilities) / sizeof(linux_capabilities[0]); i++) {
1877
+ if (linux_capabilities[i] && (cap & (1ULL << i))) {
1878
+
1879
+ if (added)
1880
+ buffer_fast_strcat(wb, " | ", 3);
1881
+
1882
+ buffer_strcat(wb, linux_capabilities[i]);
1883
+ added++;
1884
+ }
1885
+ }
1886
+ buffer_fast_strcat(wb, ")", 1);
1887
+ }
1888
+ }
1889
+}
1890
+
1891
+static void netdata_systemd_journal_transform_timestamp_usec(FACETS *facets __maybe_unused, BUFFER *wb, FACETS_TRANSFORMATION_SCOPE scope __maybe_unused, void *data __maybe_unused) {
1892
+ if(scope == FACETS_TRANSFORM_FACET_SORT)
1893
+ return;
1894
+
1895
+ const char *v = buffer_tostring(wb);
1896
+ if(*v && isdigit(*v)) {
1897
+ uint64_t ut = str2ull(buffer_tostring(wb), NULL);
1898
+ if(ut) {
1899
+ time_t timestamp_sec = ut / USEC_PER_SEC;
1900
+ struct tm tm;
1901
+ char buffer[30];
1902
+
1903
+ gmtime_r(×tamp_sec, &tm);
1904
+ strftime(buffer, sizeof(buffer), "%Y-%m-%d %H:%M:%S", &tm);
1905
+ buffer_sprintf(wb, " (%s.%06llu UTC)", buffer, ut % USEC_PER_SEC);
1906
+ }
1907
+ }
1908
+}
1909
+
1910
// ----------------------------------------------------------------------------
1911
1912
static void netdata_systemd_journal_dynamic_row_id(FACETS *facets __maybe_unused, BUFFER *json_array, FACET_ROW_KEY_VALUE *rkv, FACET_ROW *row, void *data __maybe_unused) {
1913
FACET_ROW_KEY_VALUE *pid_rkv = dictionary_get(row->dict, "_PID");
1914
const char *pid = pid_rkv ? buffer_tostring(pid_rkv->wb) : FACET_VALUE_UNSET;
1915
666
- FACET_ROW_KEY_VALUE *syslog_identifier_rkv = dictionary_get(row->dict, "SYSLOG_IDENTIFIER");
667
- const char *identifier = syslog_identifier_rkv ? buffer_tostring(syslog_identifier_rkv->wb) : FACET_VALUE_UNSET;
1916
+ const char *identifier = NULL;
1917
+ FACET_ROW_KEY_VALUE *container_name_rkv = dictionary_get(row->dict, "CONTAINER_NAME");
1918
+ if(container_name_rkv && !container_name_rkv->empty)
1919
+ identifier = buffer_tostring(container_name_rkv->wb);
1920
+
1921
+ if(!identifier) {
1922
+ FACET_ROW_KEY_VALUE *syslog_identifier_rkv = dictionary_get(row->dict, "SYSLOG_IDENTIFIER");
1923
+ if(syslog_identifier_rkv && !syslog_identifier_rkv->empty)
1924
+ identifier = buffer_tostring(syslog_identifier_rkv->wb);
1925
669
- if(strcmp(identifier, FACET_VALUE_UNSET) == 0) {
670
- FACET_ROW_KEY_VALUE *comm_rkv = dictionary_get(row->dict, "_COMM");
671
- identifier = comm_rkv ? buffer_tostring(comm_rkv->wb) : FACET_VALUE_UNSET;
1926
+ if(!identifier) {
1927
+ FACET_ROW_KEY_VALUE *comm_rkv = dictionary_get(row->dict, "_COMM");
1928
+ if(comm_rkv && !comm_rkv->empty)
1929
+ identifier = buffer_tostring(comm_rkv->wb);
1930
+ }
1931
}
1932
1933
buffer_flush(rkv->wb);
1934
676
- if(strcmp(pid, FACET_VALUE_UNSET) == 0)
677
- buffer_strcat(rkv->wb, identifier);
1935
+ if(!identifier)
1936
+ buffer_strcat(rkv->wb, FACET_VALUE_UNSET);
1937
else
1938
buffer_sprintf(rkv->wb, "%s[%s]", identifier, pid);
1939
1946
buffer_json_object_close(json_array);
1947
}
1948
1949
+DICTIONARY *function_query_status_dict = NULL;
1950
+
1951
+static void function_systemd_journal_progress(BUFFER *wb, const char *transaction, const char *progress_id) {
1952
+ if(!progress_id || !(*progress_id)) {
1953
+ netdata_mutex_lock(&stdout_mutex);
1954
+ pluginsd_function_json_error_to_stdout(transaction, HTTP_RESP_BAD_REQUEST, "missing progress id");
1955
+ netdata_mutex_unlock(&stdout_mutex);
1956
+ return;
1957
+ }
1958
+
1959
+ const DICTIONARY_ITEM *item = dictionary_get_and_acquire_item(function_query_status_dict, progress_id);
1960
+
1961
+ if(!item) {
1962
+ netdata_mutex_lock(&stdout_mutex);
1963
+ pluginsd_function_json_error_to_stdout(transaction, HTTP_RESP_NOT_FOUND, "progress id is not found here");
1964
+ netdata_mutex_unlock(&stdout_mutex);
1965
+ return;
1966
+ }
1967
+
1968
+ FUNCTION_QUERY_STATUS *fqs = dictionary_acquired_item_value(item);
1969
+
1970
+ usec_t now_monotonic_ut = now_monotonic_usec();
1971
+ if(now_monotonic_ut + 10 * USEC_PER_SEC > fqs->stop_monotonic_ut)
1972
+ fqs->stop_monotonic_ut = now_monotonic_ut + 10 * USEC_PER_SEC;
1973
+
1974
+ usec_t duration_ut = now_monotonic_ut - fqs->started_monotonic_ut;
1975
+
1976
+ size_t files_matched = fqs->files_matched;
1977
+ size_t file_working = fqs->file_working;
1978
+ if(file_working > files_matched)
1979
+ files_matched = file_working;
1980
+
1981
+ size_t rows_read = __atomic_load_n(&fqs->rows_read, __ATOMIC_RELAXED);
1982
+ size_t bytes_read = __atomic_load_n(&fqs->bytes_read, __ATOMIC_RELAXED);
1983
+
1984
+ buffer_flush(wb);
1985
+ buffer_json_initialize(wb, "\"", "\"", 0, true, BUFFER_JSON_OPTIONS_MINIFY);
1986
+ buffer_json_member_add_uint64(wb, "status", HTTP_RESP_OK);
1987
+ buffer_json_member_add_string(wb, "type", "table");
1988
+ buffer_json_member_add_uint64(wb, "running_duration_usec", duration_ut);
1989
+ buffer_json_member_add_double(wb, "progress", (double)file_working * 100.0 / (double)files_matched);
1990
+ char msg[1024 + 1];
1991
+ snprintfz(msg, 1024,
1992
+ "Read %zu rows (%0.0f rows/s), "
1993
+ "data %0.1f MB (%0.1f MB/s), "
1994
+ "file %zu of %zu",
1995
+ rows_read, (double)rows_read / (double)duration_ut * (double)USEC_PER_SEC,
1996
+ (double)bytes_read / 1024.0 / 1024.0, ((double)bytes_read / (double)duration_ut * (double)USEC_PER_SEC) / 1024.0 / 1024.0,
1997
+ file_working, files_matched
1998
+ );
1999
+ buffer_json_member_add_string(wb, "message", msg);
2000
+ buffer_json_finalize(wb);
2001
+
2002
+ netdata_mutex_lock(&stdout_mutex);
2003
+ pluginsd_function_result_to_stdout(transaction, HTTP_RESP_OK, "application/json", now_realtime_sec() + 1, wb);
2004
+ netdata_mutex_unlock(&stdout_mutex);
2005
+
2006
+ dictionary_acquired_item_release(function_query_status_dict, item);
2007
+}
2008
+
2009
static void function_systemd_journal(const char *transaction, char *function, int timeout, bool *cancelled) {
2010
+ journal_files_registry_update();
2011
+
2012
BUFFER *wb = buffer_create(0, NULL);
2013
buffer_flush(wb);
2014
buffer_json_initialize(wb, "\"", "\"", 0, true, BUFFER_JSON_OPTIONS_MINIFY);
2015
2016
+ usec_t now_monotonic_ut = now_monotonic_usec();
2017
+ FUNCTION_QUERY_STATUS tmp_fqs = {
2018
+ .cancelled = cancelled,
2019
+ .started_monotonic_ut = now_monotonic_ut,
2020
+ .stop_monotonic_ut = now_monotonic_ut + timeout * USEC_PER_SEC,
2021
+ };
2022
+ FUNCTION_QUERY_STATUS *fqs = NULL;
2023
+ const DICTIONARY_ITEM *fqs_item = NULL;
2024
+
2025
FACETS *facets = facets_create(50, FACETS_OPTION_ALL_KEYS_FTS,
2026
SYSTEMD_ALWAYS_VISIBLE_KEYS,
2027
SYSTEMD_KEYS_INCLUDED_IN_FACETS,
2039
facets_accepted_param(facets, JOURNAL_PARAMETER_HISTOGRAM);
2040
facets_accepted_param(facets, JOURNAL_PARAMETER_IF_MODIFIED_SINCE);
2041
facets_accepted_param(facets, JOURNAL_PARAMETER_DATA_ONLY);
2042
+ facets_accepted_param(facets, JOURNAL_PARAMETER_ID);
2043
+ facets_accepted_param(facets, JOURNAL_PARAMETER_PROGRESS);
2044
+ facets_accepted_param(facets, JOURNAL_PARAMETER_DELTA);
2045
+ facets_accepted_param(facets, JOURNAL_PARAMETER_TAIL);
2046
+
2047
+#ifdef HAVE_SD_JOURNAL_RESTART_FIELDS
2048
+ facets_accepted_param(facets, JOURNAL_PARAMETER_SLICE);
2049
+#endif // HAVE_SD_JOURNAL_RESTART_FIELDS
2050
2051
// register the fields in the order you want them on the dashboard
2052
2053
+ facets_register_row_severity(facets, syslog_priority_to_facet_severity, NULL);
2054
+
2055
+ facets_register_key_name(facets, "_HOSTNAME",
2056
+ FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_VISIBLE | FACET_KEY_OPTION_FTS);
2057
+
2058
facets_register_dynamic_key_name(facets, "ND_JOURNAL_PROCESS",
2059
FACET_KEY_OPTION_NEVER_FACET | FACET_KEY_OPTION_VISIBLE | FACET_KEY_OPTION_FTS,
2060
netdata_systemd_journal_dynamic_row_id, NULL);
2068
// FACET_KEY_OPTION_VISIBLE | FACET_KEY_OPTION_FTS,
2069
// netdata_systemd_journal_rich_message, NULL);
2070
728
- facets_register_key_name_transformation(facets, "PRIORITY", FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS,
2071
+ facets_register_key_name_transformation(facets, "PRIORITY",
2072
+ FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS | FACET_KEY_OPTION_TRANSFORM_VIEW,
2073
netdata_systemd_journal_transform_priority, NULL);
2074
731
- facets_register_key_name_transformation(facets, "SYSLOG_FACILITY", FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS,
2075
+ facets_register_key_name_transformation(facets, "SYSLOG_FACILITY",
2076
+ FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS | FACET_KEY_OPTION_TRANSFORM_VIEW,
2077
netdata_systemd_journal_transform_syslog_facility, NULL);
2078
734
- facets_register_key_name(facets, "SYSLOG_IDENTIFIER", FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS);
735
- facets_register_key_name(facets, "UNIT", FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS);
736
- facets_register_key_name(facets, "USER_UNIT", FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS);
2079
+ facets_register_key_name_transformation(facets, "ERRNO",
2080
+ FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS | FACET_KEY_OPTION_TRANSFORM_VIEW,
2081
+ netdata_systemd_journal_transform_errno, NULL);
2082
738
- facets_register_key_name_transformation(facets, "_UID", FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS,
2083
+ facets_register_key_name(facets, "SYSLOG_IDENTIFIER",
2084
+ FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS);
2085
+
2086
+ facets_register_key_name(facets, "UNIT",
2087
+ FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS);
2088
+
2089
+ facets_register_key_name(facets, "USER_UNIT",
2090
+ FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS);
2091
+
2092
+ facets_register_key_name_transformation(facets, "_BOOT_ID",
2093
+ FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS | FACET_KEY_OPTION_TRANSFORM_VIEW,
2094
+ netdata_systemd_journal_transform_boot_id, NULL);
2095
+
2096
+ facets_register_key_name_transformation(facets, "_SYSTEMD_OWNER_UID",
2097
+ FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS | FACET_KEY_OPTION_TRANSFORM_VIEW,
2098
netdata_systemd_journal_transform_uid, NULL);
2099
741
- facets_register_key_name_transformation(facets, "_GID", FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS,
2100
+ facets_register_key_name_transformation(facets, "_UID",
2101
+ FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS | FACET_KEY_OPTION_TRANSFORM_VIEW,
2102
+ netdata_systemd_journal_transform_uid, NULL);
2103
+
2104
+ facets_register_key_name_transformation(facets, "_GID",
2105
+ FACET_KEY_OPTION_FACET | FACET_KEY_OPTION_FTS | FACET_KEY_OPTION_TRANSFORM_VIEW,
2106
netdata_systemd_journal_transform_gid, NULL);
2107
744
- bool info = false;
745
- bool data_only = false;
2108
+ facets_register_key_name_transformation(facets, "_CAP_EFFECTIVE",
2109
+ FACET_KEY_OPTION_FTS | FACET_KEY_OPTION_TRANSFORM_VIEW,
2110
+ netdata_systemd_journal_transform_cap_effective, NULL);
2111
+
2112
+ facets_register_key_name_transformation(facets, "_AUDIT_LOGINUID",
2113
+ FACET_KEY_OPTION_FTS | FACET_KEY_OPTION_TRANSFORM_VIEW,
2114
+ netdata_systemd_journal_transform_uid, NULL);
2115
+
2116
+ facets_register_key_name_transformation(facets, "_SOURCE_REALTIME_TIMESTAMP",
2117
+ FACET_KEY_OPTION_FTS | FACET_KEY_OPTION_TRANSFORM_VIEW,
2118
+ netdata_systemd_journal_transform_timestamp_usec, NULL);
2119
+
2120
+ // ------------------------------------------------------------------------
2121
+ // parse the parameters
2122
+
2123
+ bool info = false, data_only = false, progress = false, slice = JOURNAL_DEFAULT_SLICE_MODE, delta = false, tail = false;
2124
time_t after_s = 0, before_s = 0;
2125
usec_t anchor = 0;
2126
usec_t if_modified_since = 0;
2127
size_t last = 0;
750
- FACETS_ANCHOR_DIRECTION direction = FACETS_ANCHOR_DIRECTION_BACKWARD;
2128
+ FACETS_ANCHOR_DIRECTION direction = JOURNAL_DEFAULT_DIRECTION;
2129
const char *query = NULL;
2130
const char *chart = NULL;
2131
const char *source = NULL;
2132
+ const char *progress_id = NULL;
2133
+ SD_JOURNAL_FILE_SOURCE_TYPE source_type = SDJF_ALL;
2134
+ size_t filters = 0;
2135
2136
buffer_json_member_add_object(wb, "request");
2137
2148
else if(strcmp(keyword, JOURNAL_PARAMETER_INFO) == 0) {
2149
info = true;
2150
}
770
- else if(strcmp(keyword, JOURNAL_PARAMETER_DATA_ONLY) == 0) {
771
- data_only = true;
2151
+ else if(strcmp(keyword, JOURNAL_PARAMETER_PROGRESS) == 0) {
2152
+ progress = true;
2153
+ }
2154
+ else if(strncmp(keyword, JOURNAL_PARAMETER_DELTA ":", sizeof(JOURNAL_PARAMETER_DELTA ":") - 1) == 0) {
2155
+ char *v = &keyword[sizeof(JOURNAL_PARAMETER_DELTA ":") - 1];
2156
+
2157
+ if(strcmp(v, "false") == 0 || strcmp(v, "no") == 0 || strcmp(v, "0") == 0)
2158
+ delta = false;
2159
+ else
2160
+ delta = true;
2161
+ }
2162
+ else if(strncmp(keyword, JOURNAL_PARAMETER_TAIL ":", sizeof(JOURNAL_PARAMETER_TAIL ":") - 1) == 0) {
2163
+ char *v = &keyword[sizeof(JOURNAL_PARAMETER_TAIL ":") - 1];
2164
+
2165
+ if(strcmp(v, "false") == 0 || strcmp(v, "no") == 0 || strcmp(v, "0") == 0)
2166
+ tail = false;
2167
+ else
2168
+ tail = true;
2169
+ }
2170
+ else if(strncmp(keyword, JOURNAL_PARAMETER_DATA_ONLY ":", sizeof(JOURNAL_PARAMETER_DATA_ONLY ":") - 1) == 0) {
2171
+ char *v = &keyword[sizeof(JOURNAL_PARAMETER_DATA_ONLY ":") - 1];
2172
+
2173
+ if(strcmp(v, "false") == 0 || strcmp(v, "no") == 0 || strcmp(v, "0") == 0)
2174
+ data_only = false;
2175
+ else
2176
+ data_only = true;
2177
+ }
2178
+ else if(strncmp(keyword, JOURNAL_PARAMETER_SLICE ":", sizeof(JOURNAL_PARAMETER_SLICE ":") - 1) == 0) {
2179
+ char *v = &keyword[sizeof(JOURNAL_PARAMETER_SLICE ":") - 1];
2180
+
2181
+ if(strcmp(v, "false") == 0 || strcmp(v, "no") == 0 || strcmp(v, "0") == 0)
2182
+ slice = false;
2183
+ else
2184
+ slice = true;
2185
+ }
2186
+ else if(strncmp(keyword, JOURNAL_PARAMETER_ID ":", sizeof(JOURNAL_PARAMETER_ID ":") - 1) == 0) {
2187
+ char *id = &keyword[sizeof(JOURNAL_PARAMETER_ID ":") - 1];
2188
+
2189
+ if(*id)
2190
+ progress_id = id;
2191
}
2192
else if(strncmp(keyword, JOURNAL_PARAMETER_SOURCE ":", sizeof(JOURNAL_PARAMETER_SOURCE ":") - 1) == 0) {
2193
source = &keyword[sizeof(JOURNAL_PARAMETER_SOURCE ":") - 1];
2194
+
2195
+ if(strcmp(source, SDJF_SOURCE_ALL_NAME) == 0) {
2196
+ source_type = SDJF_ALL;
2197
+ source = NULL;
2198
+ }
2199
+ else if(strcmp(source, SDJF_SOURCE_LOCAL_NAME) == 0) {
2200
+ source_type = SDJF_LOCAL;
2201
+ source = NULL;
2202
+ }
2203
+ else if(strcmp(source, SDJF_SOURCE_REMOTES_NAME) == 0) {
2204
+ source_type = SDJF_REMOTE;
2205
+ source = NULL;
2206
+ }
2207
+ else if(strcmp(source, SDJF_SOURCE_NAMESPACES_NAME) == 0) {
2208
+ source_type = SDJF_NAMESPACE;
2209
+ source = NULL;
2210
+ }
2211
+ else if(strcmp(source, SDJF_SOURCE_LOCAL_SYSTEM_NAME) == 0) {
2212
+ source_type = SDJF_LOCAL | SDJF_SYSTEM;
2213
+ source = NULL;
2214
+ }
2215
+ else if(strcmp(source, SDJF_SOURCE_LOCAL_USERS_NAME) == 0) {
2216
+ source_type = SDJF_LOCAL | SDJF_USER;
2217
+ source = NULL;
2218
+ }
2219
+ else if(strcmp(source, SDJF_SOURCE_LOCAL_OTHER_NAME) == 0) {
2220
+ source_type = SDJF_LOCAL | SDJF_OTHER;
2221
+ source = NULL;
2222
+ }
2223
+ else {
2224
+ source_type = SDJF_ALL;
2225
+ // else, match the source, whatever it is
2226
+ }
2227
}
2228
else if(strncmp(keyword, JOURNAL_PARAMETER_AFTER ":", sizeof(JOURNAL_PARAMETER_AFTER ":") - 1) == 0) {
2229
after_s = str2l(&keyword[sizeof(JOURNAL_PARAMETER_AFTER ":") - 1]);
2282
2283
facets_register_facet_id_filter(facets, keyword, value, FACET_KEY_OPTION_FACET|FACET_KEY_OPTION_FTS|FACET_KEY_OPTION_REORDER);
2284
buffer_json_add_array_item_string(wb, value);
2285
+ filters++;
2286
2287
value = sep;
2288
}
2292
}
2293
}
2294
2295
+ // ------------------------------------------------------------------------
2296
+ // put this request into the progress db
2297
+
2298
+ if(progress_id && *progress_id) {
2299
+ fqs_item = dictionary_set_and_acquire_item(function_query_status_dict, progress_id, &tmp_fqs, sizeof(tmp_fqs));
2300
+ fqs = dictionary_acquired_item_value(fqs_item);
2301
+ }
2302
+ else {
2303
+ // no progress id given, proceed without registering our progress in the dictionary
2304
+ fqs = &tmp_fqs;
2305
+ fqs_item = NULL;
2306
+ }
2307
+
2308
+ // ------------------------------------------------------------------------
2309
+ // validate parameters
2310
+
2311
time_t expires = now_realtime_sec() + 1;
2312
time_t now_s;
2313
2331
if(!last)
2332
last = SYSTEMD_JOURNAL_DEFAULT_ITEMS_PER_QUERY;
2333
865
- buffer_json_member_add_string(wb, "source", source ? source : "default");
866
- buffer_json_member_add_time_t(wb, "after", after_s);
867
- buffer_json_member_add_time_t(wb, "before", before_s);
868
- buffer_json_member_add_uint64(wb, "if_modified_since", if_modified_since);
869
- buffer_json_member_add_uint64(wb, "anchor", anchor);
870
- buffer_json_member_add_string(wb, "direction", direction == FACETS_ANCHOR_DIRECTION_FORWARD ? "forward" : "backward");
871
- buffer_json_member_add_uint64(wb, "last", last);
872
- buffer_json_member_add_string(wb, "query", query);
873
- buffer_json_member_add_string(wb, "chart", chart);
874
- buffer_json_member_add_time_t(wb, "timeout", timeout);
2334
+
2335
+ // ------------------------------------------------------------------------
2336
+ // set query time-frame, anchors and direction
2337
+
2338
+ fqs->after_ut = after_s * USEC_PER_SEC;
2339
+ fqs->before_ut = before_s * USEC_PER_SEC;
2340
+ fqs->if_modified_since = if_modified_since;
2341
+ fqs->data_only = data_only;
2342
+ fqs->delta = (fqs->data_only) ? delta : false;
2343
+ fqs->tail = (fqs->data_only && fqs->if_modified_since) ? tail : false;
2344
+ fqs->source = string_strdupz(source);
2345
+ fqs->source_type = source_type;
2346
+ fqs->entries = last;
2347
+ fqs->last_modified = 0;
2348
+ fqs->filters = filters;
2349
+ fqs->query = (query && *query) ? query : NULL;
2350
+ fqs->histogram = (chart && *chart) ? chart : NULL;
2351
+
2352
+ if(anchor && anchor < fqs->after_ut) {
2353
+ netdata_log_error("Received anchor %"PRIu64" is too small for query time-frame [%"PRIu64" - %"PRIu64"]",
2354
+ anchor, fqs->after_ut, fqs->before_ut);
2355
+ anchor = 0;
2356
+ }
2357
+ else if(anchor > fqs->before_ut) {
2358
+ netdata_log_error("Received anchor %"PRIu64" is too big for query time-frame [%"PRIu64" - %"PRIu64"]",
2359
+ anchor, fqs->after_ut, fqs->before_ut);
2360
+ anchor = 0;
2361
+ }
2362
+
2363
+ fqs->direction = direction;
2364
+ fqs->anchor.start_ut = anchor;
2365
+ fqs->anchor.stop_ut = 0;
2366
+
2367
+ if(fqs->anchor.start_ut && fqs->tail) {
2368
+ // a tail request
2369
+ // we need the top X entries from BEFORE
2370
+ // but, we need to calculate the facets and the
2371
+ // histogram up to the anchor
2372
+ fqs->direction = direction = FACETS_ANCHOR_DIRECTION_BACKWARD;
2373
+ fqs->anchor.start_ut = 0;
2374
+ fqs->anchor.stop_ut = anchor;
2375
+ }
2376
+
2377
+ facets_set_anchor(facets, fqs->anchor.start_ut, fqs->anchor.stop_ut, fqs->direction);
2378
+
2379
+ facets_set_additional_options(facets,
2380
+ ((fqs->data_only) ? FACETS_OPTION_DATA_ONLY : 0) |
2381
+ ((fqs->delta) ? FACETS_OPTION_SHOW_DELTAS : 0));
2382
+
2383
+ // ------------------------------------------------------------------------
2384
+ // set the rest of the query parameters
2385
+
2386
+
2387
+ facets_set_items(facets, fqs->entries);
2388
+ facets_set_query(facets, fqs->query);
2389
+
2390
+#ifdef HAVE_SD_JOURNAL_RESTART_FIELDS
2391
+ fqs->slice = slice;
2392
+ if(slice)
2393
+ facets_enable_slice_mode(facets);
2394
+#else
2395
+ fqs->slice = false;
2396
+#endif
2397
+
2398
+ if(fqs->histogram)
2399
+ facets_set_timeframe_and_histogram_by_id(facets, fqs->histogram, fqs->after_ut, fqs->before_ut);
2400
+ else
2401
+ facets_set_timeframe_and_histogram_by_name(facets, "PRIORITY", fqs->after_ut, fqs->before_ut);
2402
+
2403
+
2404
+ // ------------------------------------------------------------------------
2405
+ // complete the request object
2406
+
2407
+ buffer_json_member_add_boolean(wb, JOURNAL_PARAMETER_INFO, false);
2408
+ buffer_json_member_add_boolean(wb, JOURNAL_PARAMETER_SLICE, fqs->slice);
2409
+ buffer_json_member_add_boolean(wb, JOURNAL_PARAMETER_DATA_ONLY, fqs->data_only);
2410
+ buffer_json_member_add_boolean(wb, JOURNAL_PARAMETER_PROGRESS, false);
2411
+ buffer_json_member_add_boolean(wb, JOURNAL_PARAMETER_DELTA, fqs->delta);
2412
+ buffer_json_member_add_boolean(wb, JOURNAL_PARAMETER_TAIL, fqs->tail);
2413
+ buffer_json_member_add_string(wb, JOURNAL_PARAMETER_ID, progress_id);
2414
+ buffer_json_member_add_string(wb, JOURNAL_PARAMETER_SOURCE, string2str(fqs->source));
2415
+ buffer_json_member_add_uint64(wb, "source_type", fqs->source_type);
2416
+ buffer_json_member_add_uint64(wb, JOURNAL_PARAMETER_AFTER, fqs->after_ut / USEC_PER_SEC);
2417
+ buffer_json_member_add_uint64(wb, JOURNAL_PARAMETER_BEFORE, fqs->before_ut / USEC_PER_SEC);
2418
+ buffer_json_member_add_uint64(wb, "if_modified_since", fqs->if_modified_since);
2419
+ buffer_json_member_add_uint64(wb, JOURNAL_PARAMETER_ANCHOR, anchor);
2420
+ buffer_json_member_add_string(wb, JOURNAL_PARAMETER_DIRECTION, fqs->direction == FACETS_ANCHOR_DIRECTION_FORWARD ? "forward" : "backward");
2421
+ buffer_json_member_add_uint64(wb, JOURNAL_PARAMETER_LAST, fqs->entries);
2422
+ buffer_json_member_add_string(wb, JOURNAL_PARAMETER_QUERY, fqs->query);
2423
+ buffer_json_member_add_string(wb, JOURNAL_PARAMETER_HISTOGRAM, fqs->histogram);
2424
buffer_json_object_close(wb); // request
2425
2426
+ buffer_json_journal_versions(wb);
2427
+
2428
+ // ------------------------------------------------------------------------
2429
+ // run the request
2430
+
2431
int response;
2432
2433
if(info) {
2442
buffer_json_member_add_string(wb, "type", "select");
2443
buffer_json_member_add_array(wb, "options");
2444
{
891
- buffer_json_add_array_item_object(wb);
892
- {
893
- buffer_json_member_add_string(wb, "id", "default");
894
- buffer_json_member_add_string(wb, "name", "default");
895
- }
896
- buffer_json_object_close(wb); // options object
2445
+ available_journal_file_sources_to_json_array(wb);
2446
}
2447
buffer_json_array_close(wb); // options array
2448
}
2450
}
2451
buffer_json_array_close(wb); // required_params array
2452
2453
+ facets_table_config(wb);
2454
+
2455
buffer_json_member_add_uint64(wb, "status", HTTP_RESP_OK);
2456
buffer_json_member_add_string(wb, "type", "table");
2457
buffer_json_member_add_string(wb, "help", SYSTEMD_JOURNAL_FUNCTION_DESCRIPTION);
2460
goto output;
2461
}
2462
912
- facets_set_items(facets, last);
913
- facets_set_anchor(facets, anchor, direction);
914
- facets_set_query(facets, query);
2463
+ if(progress) {
2464
+ function_systemd_journal_progress(wb, transaction, progress_id);
2465
+ goto cleanup;
2466
+ }
2467
916
- if(chart && *chart)
917
- facets_set_histogram_by_id(facets, chart,
918
- after_s * USEC_PER_SEC, before_s * USEC_PER_SEC);
919
- else
920
- facets_set_histogram_by_name(facets, "PRIORITY",
921
- after_s * USEC_PER_SEC, before_s * USEC_PER_SEC);
2468
+ response = netdata_systemd_journal_query(wb, facets, fqs);
2469
+
2470
+ // ------------------------------------------------------------------------
2471
+ // cleanup query params
2472
+
2473
+ string_freez(fqs->source);
2474
+ fqs->source = NULL;
2475
923
- response = netdata_systemd_journal_query(wb, facets, after_s * USEC_PER_SEC, before_s * USEC_PER_SEC,
924
- anchor, direction, last,
925
- if_modified_since, data_only,
926
- now_monotonic_usec() + (timeout - 1) * USEC_PER_SEC,
927
- cancelled);
2476
+ // ------------------------------------------------------------------------
2477
+ // handle error response
2478
2479
if(response != HTTP_RESP_OK) {
2480
netdata_mutex_lock(&stdout_mutex);
2491
cleanup:
2492
facets_destroy(facets);
2493
buffer_free(wb);
2494
+
2495
+ if(fqs_item) {
2496
+ dictionary_del(function_query_status_dict, dictionary_acquired_item_name(fqs_item));
2497
+ dictionary_acquired_item_release(function_query_status_dict, fqs_item);
2498
+ dictionary_garbage_collect(function_query_status_dict);
2499
+ }
2500
}
2501
2502
// ----------------------------------------------------------------------------
2517
netdata_configured_host_prefix = getenv("NETDATA_HOST_PREFIX");
2518
if(verify_netdata_host_prefix() == -1) exit(1);
2519
2520
+ // ------------------------------------------------------------------------
2521
+ // setup the journal directories
2522
+
2523
+ unsigned d = 0;
2524
+
2525
+ journal_directories[d++].path = strdupz("/var/log/journal");
2526
+ journal_directories[d++].path = strdupz("/run/log/journal");
2527
+
2528
+ if(*netdata_configured_host_prefix) {
2529
+ char path[PATH_MAX];
2530
+ snprintfz(path, sizeof(path), "%s/var/log/journal", netdata_configured_host_prefix);
2531
+ journal_directories[d++].path = strdupz(path);
2532
+ snprintfz(path, sizeof(path), "%s/run/log/journal", netdata_configured_host_prefix);
2533
+ journal_directories[d++].path = strdupz(path);
2534
+ }
2535
+
2536
+ // terminate the list
2537
+ journal_directories[d].path = NULL;
2538
+
2539
+ // ------------------------------------------------------------------------
2540
+
2541
+ function_query_status_dict = dictionary_create_advanced(
2542
+ DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_FIXED_SIZE,
2543
+ NULL, sizeof(FUNCTION_QUERY_STATUS));
2544
+
2545
+ // ------------------------------------------------------------------------
2546
+ // initialize the used hashes files registry
2547
+
2548
+ used_hashes_registry = dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE);
2549
+
2550
+
2551
+ // ------------------------------------------------------------------------
2552
+ // initialize the journal files registry
2553
+
2554
+ systemd_journal_session = (now_realtime_usec() / USEC_PER_SEC) * USEC_PER_SEC;
2555
+
2556
+ journal_files_registry = dictionary_create_advanced(
2557
+ DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_FIXED_SIZE,
2558
+ NULL, sizeof(struct journal_file));
2559
+
2560
+ dictionary_register_insert_callback(journal_files_registry, files_registry_insert_cb, NULL);
2561
+ dictionary_register_delete_callback(journal_files_registry, files_registry_delete_cb, NULL);
2562
+ dictionary_register_conflict_callback(journal_files_registry, files_registry_conflict_cb, NULL);
2563
+
2564
+ boot_ids_to_first_ut = dictionary_create_advanced(
2565
+ DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_FIXED_SIZE,
2566
+ NULL, sizeof(usec_t));
2567
+
2568
+ journal_files_registry_update();
2569
+
2570
+
2571
// ------------------------------------------------------------------------
2572
// debug
2573
2574
if(argc == 2 && strcmp(argv[1], "debug") == 0) {
2575
bool cancelled = false;
2576
char buf[] = "systemd-journal after:-2592000 before:0 last:500";
2577
+ // char buf[] = "systemd-journal after:1695332964 before:1695937764 direction:backward last:100 slice:true source:all DHKucpqUoe1:PtVoyIuX.MU";
2578
// char buf[] = "systemd-journal after:1694511062 before:1694514662 anchor:1694514122024403";
971
- function_systemd_journal("123", buf, 30, &cancelled);
2579
+ function_systemd_journal("123", buf, 600, &cancelled);
2580
exit(1);
2581
}
2582