| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #ifndef NETDATA_SYSTEMD_JOURNAL_EXECUTE_H |
| 4 | #define NETDATA_SYSTEMD_JOURNAL_EXECUTE_H |
| 5 | |
| 6 | #define JOURNAL_KEY_ND_JOURNAL_FILE "ND_JOURNAL_FILE" |
| 7 | |
| 8 | #define FUNCTION_PROGRESS_UPDATE_ROWS(rows_read, rows) __atomic_fetch_add(&(rows_read), rows, __ATOMIC_RELAXED) |
| 9 | #define FUNCTION_PROGRESS_UPDATE_BYTES(bytes_read, bytes) __atomic_fetch_add(&(bytes_read), bytes, __ATOMIC_RELAXED) |
| 10 | #define FUNCTION_PROGRESS_EVERY_ROWS (1ULL << 13) |
| 11 | #define FUNCTION_DATA_ONLY_CHECK_EVERY_ROWS (1ULL << 7) |
| 12 | |
| 13 | #include "systemd-journal-sampling.h" |
| 14 | |
| 15 | static __always_inline |
| 16 | bool nd_sd_journal_seek_to(NsdJournal *j, usec_t timestamp) |
| 17 | { |
| 18 | if (nsd_journal_seek_realtime_usec(j, timestamp) < 0) { |
| 19 | netdata_log_error("SYSTEMD-JOURNAL: Failed to seek to %" PRIu64, timestamp); |
| 20 | if (nsd_journal_seek_tail(j) < 0) { |
| 21 | netdata_log_error("SYSTEMD-JOURNAL: Failed to seek to journal's tail"); |
| 22 | return false; |
| 23 | } |
| 24 | } |
| 25 | |
| 26 | return true; |
| 27 | } |
| 28 | |
| 29 | static __always_inline |
| 30 | size_t nd_sd_journal_process_row(NsdJournal *j, FACETS *facets, struct nd_journal_file *njf, usec_t *msg_ut) { |
| 31 | const void *data; |
| 32 | size_t length, bytes = 0; |
| 33 | |
| 34 | facets_add_key_value_length( |
| 35 | facets, JOURNAL_KEY_ND_JOURNAL_FILE, sizeof(JOURNAL_KEY_ND_JOURNAL_FILE) - 1, njf->filename, njf->filename_len); |
| 36 | |
| 37 | NSD_JOURNAL_FOREACH_DATA(j, data, length) |
| 38 | { |
| 39 | const char *key, *value; |
| 40 | size_t key_length, value_length; |
| 41 | |
| 42 | if (!parse_journal_field(data, length, &key, &key_length, &value, &value_length)) |
| 43 | continue; |
| 44 | |
| 45 | #ifdef NETDATA_INTERNAL_CHECKS |
| 46 | usec_t origin_journal_ut = *msg_ut; |
| 47 | #endif |
| 48 | if (unlikely( |
| 49 | key_length == sizeof(JD_SOURCE_REALTIME_TIMESTAMP) - 1 && |
| 50 | memcmp(key, JD_SOURCE_REALTIME_TIMESTAMP, sizeof(JD_SOURCE_REALTIME_TIMESTAMP) - 1) == 0)) { |
| 51 | usec_t ut = str2ull(value, NULL); |
| 52 | if (ut && ut < *msg_ut) { |
| 53 | usec_t delta = *msg_ut - ut; |
| 54 | *msg_ut = ut; |
| 55 | |
| 56 | if (delta > JOURNAL_VS_REALTIME_DELTA_MAX_UT) |
| 57 | delta = JOURNAL_VS_REALTIME_DELTA_MAX_UT; |
| 58 | |
| 59 | // update max_journal_vs_realtime_delta_ut if the delta increased |
| 60 | usec_t expected = njf->max_journal_vs_realtime_delta_ut; |
| 61 | do { |
| 62 | if (delta <= expected) |
| 63 | break; |
| 64 | } while (!__atomic_compare_exchange_n( |
| 65 | &njf->max_journal_vs_realtime_delta_ut, &expected, delta, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED)); |
| 66 | |
| 67 | internal_error( |
| 68 | delta > expected, |
| 69 | "increased max_journal_vs_realtime_delta_ut from %" PRIu64 " to %" PRIu64 ", " |
| 70 | "journal %" PRIu64 ", actual %" PRIu64 " (delta %" PRIu64 ")", |
| 71 | expected, |
| 72 | delta, |
| 73 | origin_journal_ut, |
| 74 | *msg_ut, |
| 75 | origin_journal_ut - (*msg_ut)); |
| 76 | } |
| 77 | } |
| 78 | |
| 79 | bytes += length; |
| 80 | facets_add_key_value_length( |
| 81 | facets, |
| 82 | key, |
| 83 | key_length, |
| 84 | value, |
| 85 | value_length <= FACET_MAX_VALUE_LENGTH ? value_length : FACET_MAX_VALUE_LENGTH); |
| 86 | } |
| 87 | |
| 88 | return bytes; |
| 89 | } |
| 90 | |
| 91 | static __always_inline |
| 92 | ND_SD_JOURNAL_STATUS check_stop(const bool *cancelled, const usec_t *stop_monotonic_ut) { |
| 93 | if (cancelled && __atomic_load_n(cancelled, __ATOMIC_RELAXED)) { |
| 94 | internal_error(true, "Function has been cancelled"); |
| 95 | return ND_SD_JOURNAL_CANCELLED; |
| 96 | } |
| 97 | |
| 98 | if (now_monotonic_usec() > __atomic_load_n(stop_monotonic_ut, __ATOMIC_RELAXED)) { |
| 99 | internal_error(true, "Function timed out"); |
| 100 | return ND_SD_JOURNAL_TIMED_OUT; |
| 101 | } |
| 102 | |
| 103 | return ND_SD_JOURNAL_OK; |
| 104 | } |
| 105 | |
| 106 | static __always_inline |
| 107 | ND_SD_JOURNAL_STATUS nd_sd_journal_query_backward( |
| 108 | NsdJournal *j, |
| 109 | BUFFER *wb __maybe_unused, |
| 110 | FACETS *facets, |
| 111 | struct nd_journal_file *njf, |
| 112 | LOGS_QUERY_STATUS *fqs) |
| 113 | { |
| 114 | usec_t anchor_delta = __atomic_load_n(&njf->max_journal_vs_realtime_delta_ut, __ATOMIC_RELAXED); |
| 115 | lqs_query_timeframe(fqs, anchor_delta); |
| 116 | usec_t start_ut = fqs->query.start_ut; |
| 117 | usec_t stop_ut = fqs->query.stop_ut; |
| 118 | bool stop_when_full = fqs->query.stop_when_full; |
| 119 | |
| 120 | fqs->c.query_file.start_ut = start_ut; |
| 121 | fqs->c.query_file.stop_ut = stop_ut; |
| 122 | |
| 123 | if (!nd_sd_journal_seek_to(j, start_ut)) |
| 124 | return ND_SD_JOURNAL_FAILED_TO_SEEK; |
| 125 | |
| 126 | size_t errors_no_timestamp = 0; |
| 127 | usec_t latest_msg_ut = 0; // the biggest timestamp we have seen so far |
| 128 | usec_t first_msg_ut = 0; // the first message we got from the db |
| 129 | size_t row_counter = 0, last_row_counter = 0, rows_useful = 0; |
| 130 | size_t bytes = 0, last_bytes = 0; |
| 131 | |
| 132 | usec_t last_usec_from = 0; |
| 133 | usec_t last_usec_to = 0; |
| 134 | |
| 135 | ND_SD_JOURNAL_STATUS status = ND_SD_JOURNAL_OK; |
| 136 | |
| 137 | facets_rows_begin(facets); |
| 138 | while (status == ND_SD_JOURNAL_OK && nsd_journal_previous(j) > 0) { |
| 139 | usec_t msg_ut = 0; |
| 140 | if (nsd_journal_get_realtime_usec(j, &msg_ut) < 0 || !msg_ut) { |
| 141 | errors_no_timestamp++; |
| 142 | continue; |
| 143 | } |
| 144 | |
| 145 | if (unlikely(msg_ut > start_ut)) |
| 146 | continue; |
| 147 | |
| 148 | if (unlikely(msg_ut < stop_ut)) |
| 149 | break; |
| 150 | |
| 151 | if (unlikely(msg_ut > latest_msg_ut)) |
| 152 | latest_msg_ut = msg_ut; |
| 153 | |
| 154 | if (unlikely(!first_msg_ut)) { |
| 155 | first_msg_ut = msg_ut; |
| 156 | fqs->c.query_file.first_msg_ut = msg_ut; |
| 157 | |
| 158 | #ifdef HAVE_SD_JOURNAL_GET_SEQNUM |
| 159 | if (nsd_journal_get_seqnum(j, &fqs->c.query_file.first_msg_seqnum, &fqs->c.query_file.first_msg_writer) < |
| 160 | 0) { |
| 161 | fqs->c.query_file.first_msg_seqnum = 0; |
| 162 | fqs->c.query_file.first_msg_writer = NSD_ID128_NULL; |
| 163 | } |
| 164 | #endif |
| 165 | } |
| 166 | |
| 167 | sampling_t sample = is_row_in_sample( |
| 168 | j, fqs, njf, msg_ut, FACETS_ANCHOR_DIRECTION_BACKWARD, facets_row_candidate_to_keep(facets, msg_ut)); |
| 169 | |
| 170 | if (sample == SAMPLING_FULL) { |
| 171 | bytes += nd_sd_journal_process_row(j, facets, njf, &msg_ut); |
| 172 | |
| 173 | // make sure each line gets a unique timestamp |
| 174 | if (unlikely(msg_ut >= last_usec_from && msg_ut <= last_usec_to)) |
| 175 | msg_ut = --last_usec_from; |
| 176 | else |
| 177 | last_usec_from = last_usec_to = msg_ut; |
| 178 | |
| 179 | if (facets_row_finished(facets, msg_ut)) |
| 180 | rows_useful++; |
| 181 | |
| 182 | row_counter++; |
| 183 | if (unlikely( |
| 184 | (row_counter % FUNCTION_DATA_ONLY_CHECK_EVERY_ROWS) == 0 && stop_when_full && |
| 185 | facets_rows(facets) >= fqs->rq.entries)) { |
| 186 | // stop the data only query |
| 187 | usec_t oldest = facets_row_oldest_ut(facets); |
| 188 | if (oldest && msg_ut < (oldest - anchor_delta)) |
| 189 | break; |
| 190 | } |
| 191 | |
| 192 | if (unlikely(row_counter % FUNCTION_PROGRESS_EVERY_ROWS == 0)) { |
| 193 | FUNCTION_PROGRESS_UPDATE_ROWS(fqs->c.rows_read, row_counter - last_row_counter); |
| 194 | last_row_counter = row_counter; |
| 195 | |
| 196 | FUNCTION_PROGRESS_UPDATE_BYTES(fqs->c.bytes_read, bytes - last_bytes); |
| 197 | last_bytes = bytes; |
| 198 | |
| 199 | status = check_stop(fqs->cancelled, fqs->stop_monotonic_ut); |
| 200 | } |
| 201 | } else if (sample == SAMPLING_SKIP_FIELDS) |
| 202 | facets_row_finished_unsampled(facets, msg_ut); |
| 203 | else { |
| 204 | sampling_update_running_query_file_estimates(facets, j, fqs, njf, msg_ut, FACETS_ANCHOR_DIRECTION_BACKWARD); |
| 205 | break; |
| 206 | } |
| 207 | } |
| 208 | |
| 209 | FUNCTION_PROGRESS_UPDATE_ROWS(fqs->c.rows_read, row_counter - last_row_counter); |
| 210 | FUNCTION_PROGRESS_UPDATE_BYTES(fqs->c.bytes_read, bytes - last_bytes); |
| 211 | |
| 212 | fqs->c.rows_useful += rows_useful; |
| 213 | |
| 214 | if (errors_no_timestamp) |
| 215 | netdata_log_error("SYSTEMD-JOURNAL: %zu lines did not have timestamps", errors_no_timestamp); |
| 216 | |
| 217 | if (latest_msg_ut > fqs->last_modified) |
| 218 | fqs->last_modified = latest_msg_ut; |
| 219 | |
| 220 | return status; |
| 221 | } |
| 222 | |
| 223 | static __always_inline |
| 224 | ND_SD_JOURNAL_STATUS nd_sd_journal_query_forward( |
| 225 | NsdJournal *j, |
| 226 | BUFFER *wb __maybe_unused, |
| 227 | FACETS *facets, |
| 228 | struct nd_journal_file *njf, |
| 229 | LOGS_QUERY_STATUS *fqs) |
| 230 | { |
| 231 | usec_t anchor_delta = __atomic_load_n(&njf->max_journal_vs_realtime_delta_ut, __ATOMIC_RELAXED); |
| 232 | lqs_query_timeframe(fqs, anchor_delta); |
| 233 | usec_t start_ut = fqs->query.start_ut; |
| 234 | usec_t stop_ut = fqs->query.stop_ut; |
| 235 | bool stop_when_full = fqs->query.stop_when_full; |
| 236 | |
| 237 | fqs->c.query_file.start_ut = start_ut; |
| 238 | fqs->c.query_file.stop_ut = stop_ut; |
| 239 | |
| 240 | if (!nd_sd_journal_seek_to(j, start_ut)) |
| 241 | return ND_SD_JOURNAL_FAILED_TO_SEEK; |
| 242 | |
| 243 | size_t errors_no_timestamp = 0; |
| 244 | usec_t latest_msg_ut = 0; // the biggest timestamp we have seen so far |
| 245 | usec_t first_msg_ut = 0; // the first message we got from the db |
| 246 | size_t row_counter = 0, last_row_counter = 0, rows_useful = 0; |
| 247 | size_t bytes = 0, last_bytes = 0; |
| 248 | |
| 249 | usec_t last_usec_from = 0; |
| 250 | usec_t last_usec_to = 0; |
| 251 | |
| 252 | ND_SD_JOURNAL_STATUS status = ND_SD_JOURNAL_OK; |
| 253 | |
| 254 | facets_rows_begin(facets); |
| 255 | while (status == ND_SD_JOURNAL_OK && nsd_journal_next(j) > 0) { |
| 256 | usec_t msg_ut = 0; |
| 257 | if (nsd_journal_get_realtime_usec(j, &msg_ut) < 0 || !msg_ut) { |
| 258 | errors_no_timestamp++; |
| 259 | continue; |
| 260 | } |
| 261 | |
| 262 | if (unlikely(msg_ut < start_ut)) |
| 263 | continue; |
| 264 | |
| 265 | if (unlikely(msg_ut > stop_ut)) |
| 266 | break; |
| 267 | |
| 268 | if (likely(msg_ut > latest_msg_ut)) |
| 269 | latest_msg_ut = msg_ut; |
| 270 | |
| 271 | if (unlikely(!first_msg_ut)) { |
| 272 | first_msg_ut = msg_ut; |
| 273 | fqs->c.query_file.first_msg_ut = msg_ut; |
| 274 | } |
| 275 | |
| 276 | sampling_t sample = is_row_in_sample( |
| 277 | j, fqs, njf, msg_ut, FACETS_ANCHOR_DIRECTION_FORWARD, facets_row_candidate_to_keep(facets, msg_ut)); |
| 278 | |
| 279 | if (sample == SAMPLING_FULL) { |
| 280 | bytes += nd_sd_journal_process_row(j, facets, njf, &msg_ut); |
| 281 | |
| 282 | // make sure each line gets a unique timestamp |
| 283 | if (unlikely(msg_ut >= last_usec_from && msg_ut <= last_usec_to)) |
| 284 | msg_ut = ++last_usec_to; |
| 285 | else |
| 286 | last_usec_from = last_usec_to = msg_ut; |
| 287 | |
| 288 | if (facets_row_finished(facets, msg_ut)) |
| 289 | rows_useful++; |
| 290 | |
| 291 | row_counter++; |
| 292 | if (unlikely( |
| 293 | (row_counter % FUNCTION_DATA_ONLY_CHECK_EVERY_ROWS) == 0 && stop_when_full && |
| 294 | facets_rows(facets) >= fqs->rq.entries)) { |
| 295 | // stop the data only query |
| 296 | usec_t newest = facets_row_newest_ut(facets); |
| 297 | if (newest && msg_ut > (newest + anchor_delta)) |
| 298 | break; |
| 299 | } |
| 300 | |
| 301 | if (unlikely(row_counter % FUNCTION_PROGRESS_EVERY_ROWS == 0)) { |
| 302 | FUNCTION_PROGRESS_UPDATE_ROWS(fqs->c.rows_read, row_counter - last_row_counter); |
| 303 | last_row_counter = row_counter; |
| 304 | |
| 305 | FUNCTION_PROGRESS_UPDATE_BYTES(fqs->c.bytes_read, bytes - last_bytes); |
| 306 | last_bytes = bytes; |
| 307 | |
| 308 | status = check_stop(fqs->cancelled, fqs->stop_monotonic_ut); |
| 309 | } |
| 310 | } else if (sample == SAMPLING_SKIP_FIELDS) |
| 311 | facets_row_finished_unsampled(facets, msg_ut); |
| 312 | else { |
| 313 | sampling_update_running_query_file_estimates(facets, j, fqs, njf, msg_ut, FACETS_ANCHOR_DIRECTION_FORWARD); |
| 314 | break; |
| 315 | } |
| 316 | } |
| 317 | |
| 318 | FUNCTION_PROGRESS_UPDATE_ROWS(fqs->c.rows_read, row_counter - last_row_counter); |
| 319 | FUNCTION_PROGRESS_UPDATE_BYTES(fqs->c.bytes_read, bytes - last_bytes); |
| 320 | |
| 321 | fqs->c.rows_useful += rows_useful; |
| 322 | |
| 323 | if (errors_no_timestamp) |
| 324 | netdata_log_error("SYSTEMD-JOURNAL: %zu lines did not have timestamps", errors_no_timestamp); |
| 325 | |
| 326 | if (latest_msg_ut > fqs->last_modified) |
| 327 | fqs->last_modified = latest_msg_ut; |
| 328 | |
| 329 | return status; |
| 330 | } |
| 331 | |
| 332 | bool nd_sd_journal_check_if_modified_since(NsdJournal *j, usec_t seek_to, usec_t last_modified) |
| 333 | { |
| 334 | // return true, if data have been modified since the timestamp |
| 335 | |
| 336 | if (!last_modified || !seek_to) |
| 337 | return false; |
| 338 | |
| 339 | if (!nd_sd_journal_seek_to(j, seek_to)) |
| 340 | return false; |
| 341 | |
| 342 | usec_t first_msg_ut = 0; |
| 343 | while (nsd_journal_previous(j) > 0) { |
| 344 | usec_t msg_ut; |
| 345 | if (nsd_journal_get_realtime_usec(j, &msg_ut) < 0) |
| 346 | continue; |
| 347 | |
| 348 | first_msg_ut = msg_ut; |
| 349 | break; |
| 350 | } |
| 351 | |
| 352 | return first_msg_ut != last_modified; |
| 353 | } |
| 354 | |
| 355 | #ifdef HAVE_SD_JOURNAL_RESTART_FIELDS |
| 356 | static __always_inline |
| 357 | bool netdata_systemd_filtering_by_journal(NsdJournal *j, FACETS *facets, LOGS_QUERY_STATUS *lqs) { |
| 358 | const char *field = NULL; |
| 359 | const void *data = NULL; |
| 360 | size_t data_length; |
| 361 | size_t added_keys = 0; |
| 362 | size_t failures = 0; |
| 363 | size_t filters_added = 0; |
| 364 | |
| 365 | NSD_JOURNAL_FOREACH_FIELD(j, field) |
| 366 | { // for each key |
| 367 | bool interesting; |
| 368 | |
| 369 | if (lqs->rq.data_only) |
| 370 | interesting = facets_key_name_is_filter(facets, field); |
| 371 | else |
| 372 | interesting = facets_key_name_is_facet(facets, field); |
| 373 | |
| 374 | if (interesting) { |
| 375 | int r = nsd_journal_query_unique(j, field); |
| 376 | if (r >= 0) { |
| 377 | bool added_this_key = false; |
| 378 | size_t added_values = 0; |
| 379 | |
| 380 | nsd_journal_restart_unique(j); |
| 381 | while ((r = nsd_journal_enumerate_available_unique(j, &data, &data_length)) > 0) { |
| 382 | // for each value of the key |
| 383 | const char *key, *value; |
| 384 | size_t key_length, value_length; |
| 385 | |
| 386 | if (!parse_journal_field(data, data_length, &key, &key_length, &value, &value_length)) |
| 387 | continue; |
| 388 | |
| 389 | facets_add_possible_value_name_to_key(facets, key, key_length, value, value_length); |
| 390 | |
| 391 | if (!facets_key_name_value_length_is_selected(facets, key, key_length, value, value_length)) |
| 392 | continue; |
| 393 | |
| 394 | if (added_keys && !added_this_key) { |
| 395 | if (nsd_journal_add_conjunction(j) < 0) // key AND key AND key |
| 396 | failures++; |
| 397 | |
| 398 | added_this_key = true; |
| 399 | added_keys++; |
| 400 | } else if (added_values) |
| 401 | if (nsd_journal_add_disjunction(j) < 0) // value OR value OR value |
| 402 | failures++; |
| 403 | |
| 404 | if (nsd_journal_add_match(j, data, data_length) < 0) |
| 405 | failures++; |
| 406 | |
| 407 | if (!added_keys) { |
| 408 | added_keys++; |
| 409 | added_this_key = true; |
| 410 | } |
| 411 | |
| 412 | added_values++; |
| 413 | filters_added++; |
| 414 | } |
| 415 | |
| 416 | if (r < 0) |
| 417 | failures++; |
| 418 | } else { |
| 419 | failures++; |
| 420 | } |
| 421 | } |
| 422 | } |
| 423 | |
| 424 | if (failures) { |
| 425 | lqs_log_error(lqs, "failed to setup journal filter, will run the full query."); |
| 426 | nsd_journal_flush_matches(j); |
| 427 | return true; |
| 428 | } |
| 429 | |
| 430 | return filters_added ? true : false; |
| 431 | } |
| 432 | #endif // HAVE_SD_JOURNAL_RESTART_FIELDS |
| 433 | |
| 434 | static __always_inline |
| 435 | ND_SD_JOURNAL_STATUS nd_sd_journal_query_one_file( |
| 436 | const char *filename, |
| 437 | BUFFER *wb, |
| 438 | FACETS *facets, |
| 439 | struct nd_journal_file *njf, |
| 440 | LOGS_QUERY_STATUS *fqs) |
| 441 | { |
| 442 | NsdJournal *j = NULL; |
| 443 | errno_clear(); |
| 444 | |
| 445 | fstat_cache_enable_on_thread(); |
| 446 | |
| 447 | const char *paths[2] = { |
| 448 | [0] = filename, |
| 449 | [1] = NULL, |
| 450 | }; |
| 451 | |
| 452 | if (nsd_journal_open_files(&j, paths, ND_SD_JOURNAL_OPEN_FLAGS) < 0 || !j) { |
| 453 | netdata_log_error("JOURNAL: cannot open file '%s' for query", filename); |
| 454 | fstat_cache_disable_on_thread(); |
| 455 | return ND_SD_JOURNAL_FAILED_TO_OPEN; |
| 456 | } |
| 457 | |
| 458 | ND_SD_JOURNAL_STATUS status; |
| 459 | bool matches_filters = true; |
| 460 | |
| 461 | #ifdef HAVE_SD_JOURNAL_RESTART_FIELDS |
| 462 | if (fqs->rq.slice) { |
| 463 | usec_t started = now_monotonic_usec(); |
| 464 | |
| 465 | matches_filters = netdata_systemd_filtering_by_journal(j, facets, fqs) || !fqs->rq.filters; |
| 466 | usec_t ended = now_monotonic_usec(); |
| 467 | |
| 468 | fqs->c.matches_setup_ut += (ended - started); |
| 469 | } |
| 470 | #endif // HAVE_SD_JOURNAL_RESTART_FIELDS |
| 471 | |
| 472 | if (matches_filters) { |
| 473 | if (fqs->rq.direction == FACETS_ANCHOR_DIRECTION_FORWARD) |
| 474 | status = nd_sd_journal_query_forward(j, wb, facets, njf, fqs); |
| 475 | else |
| 476 | status = nd_sd_journal_query_backward(j, wb, facets, njf, fqs); |
| 477 | } else |
| 478 | status = ND_SD_JOURNAL_NO_FILE_MATCHED; |
| 479 | |
| 480 | nsd_journal_close(j); |
| 481 | fstat_cache_disable_on_thread(); |
| 482 | |
| 483 | return status; |
| 484 | } |
| 485 | |
| 486 | static __always_inline |
| 487 | bool jf_is_mine(struct nd_journal_file *njf, LOGS_QUERY_STATUS *fqs) { |
| 488 | if ((fqs->rq.source_type == ND_SD_JF_NONE && !fqs->rq.sources) || (njf->source_type & fqs->rq.source_type) || |
| 489 | (fqs->rq.sources && simple_pattern_matches(fqs->rq.sources, string2str(njf->source)))) { |
| 490 | if (!njf->msg_last_ut) |
| 491 | // the file is not scanned yet, or the timestamps have not been updated, |
| 492 | // so we don't know if it can contribute or not - let's add it. |
| 493 | return true; |
| 494 | |
| 495 | usec_t anchor_delta = JOURNAL_VS_REALTIME_DELTA_MAX_UT; |
| 496 | usec_t first_ut = njf->msg_first_ut - anchor_delta; |
| 497 | usec_t last_ut = njf->msg_last_ut + anchor_delta; |
| 498 | |
| 499 | if (last_ut >= fqs->rq.after_ut && first_ut <= fqs->rq.before_ut) |
| 500 | return true; |
| 501 | } |
| 502 | |
| 503 | return false; |
| 504 | } |
| 505 | |
| 506 | static __always_inline |
| 507 | int nd_sd_journal_query(BUFFER *wb, LOGS_QUERY_STATUS *lqs) { |
| 508 | FACETS *facets = lqs->facets; |
| 509 | |
| 510 | ND_SD_JOURNAL_STATUS status = ND_SD_JOURNAL_NO_FILE_MATCHED; |
| 511 | struct nd_journal_file *njf; |
| 512 | |
| 513 | lqs->c.files_matched = 0; |
| 514 | lqs->c.file_working = 0; |
| 515 | lqs->c.rows_useful = 0; |
| 516 | lqs->c.rows_read = 0; |
| 517 | lqs->c.bytes_read = 0; |
| 518 | |
| 519 | size_t files_used = 0; |
| 520 | size_t files_max = dictionary_entries(nd_journal_files_registry); |
| 521 | const DICTIONARY_ITEM *file_items[files_max]; |
| 522 | |
| 523 | // count the files |
| 524 | bool files_are_newer = false; |
| 525 | dfe_start_read(nd_journal_files_registry, njf) |
| 526 | { |
| 527 | if (!jf_is_mine(njf, lqs)) |
| 528 | continue; |
| 529 | |
| 530 | file_items[files_used++] = dictionary_acquired_item_dup(nd_journal_files_registry, njf_dfe.item); |
| 531 | |
| 532 | if (njf->msg_last_ut > lqs->rq.if_modified_since) |
| 533 | files_are_newer = true; |
| 534 | } |
| 535 | dfe_done(jf); |
| 536 | |
| 537 | lqs->c.files_matched = files_used; |
| 538 | |
| 539 | if (lqs->rq.if_modified_since && !files_are_newer) { |
| 540 | // release the files |
| 541 | for (size_t f = 0; f < files_used; f++) |
| 542 | dictionary_acquired_item_release(nd_journal_files_registry, file_items[f]); |
| 543 | |
| 544 | return rrd_call_function_error(wb, "No new data since the previous call.", HTTP_RESP_NOT_MODIFIED); |
| 545 | } |
| 546 | |
| 547 | // sort the files, so that they are optimal for facets |
| 548 | if (files_used >= 2) { |
| 549 | if (lqs->rq.direction == FACETS_ANCHOR_DIRECTION_BACKWARD) |
| 550 | qsort(file_items, files_used, sizeof(const DICTIONARY_ITEM *), nd_journal_file_dict_items_backward_compar); |
| 551 | else |
| 552 | qsort(file_items, files_used, sizeof(const DICTIONARY_ITEM *), nd_journal_file_dict_items_forward_compar); |
| 553 | } |
| 554 | |
| 555 | bool partial = false; |
| 556 | usec_t query_started_ut = now_monotonic_usec(); |
| 557 | usec_t started_ut = query_started_ut; |
| 558 | usec_t ended_ut = started_ut; |
| 559 | usec_t duration_ut = 0, max_duration_ut = 0; |
| 560 | usec_t progress_duration_ut = 0; |
| 561 | |
| 562 | sampling_query_init(lqs, facets); |
| 563 | |
| 564 | buffer_json_member_add_array(wb, "_journal_files"); |
| 565 | for (size_t f = 0; f < files_used; f++) { |
| 566 | const char *filename = dictionary_acquired_item_name(file_items[f]); |
| 567 | njf = dictionary_acquired_item_value(file_items[f]); |
| 568 | |
| 569 | if (!jf_is_mine(njf, lqs)) |
| 570 | continue; |
| 571 | |
| 572 | started_ut = ended_ut; |
| 573 | |
| 574 | // do not even try to do the query if we expect it to pass the timeout |
| 575 | if (ended_ut + max_duration_ut * 3 >= *lqs->stop_monotonic_ut) { |
| 576 | partial = true; |
| 577 | status = ND_SD_JOURNAL_TIMED_OUT; |
| 578 | break; |
| 579 | } |
| 580 | |
| 581 | lqs->c.file_working++; |
| 582 | |
| 583 | size_t fs_calls = fstat_thread_calls; |
| 584 | size_t fs_cached = fstat_thread_cached_responses; |
| 585 | size_t rows_useful = lqs->c.rows_useful; |
| 586 | size_t rows_read = lqs->c.rows_read; |
| 587 | size_t bytes_read = lqs->c.bytes_read; |
| 588 | size_t matches_setup_ut = lqs->c.matches_setup_ut; |
| 589 | |
| 590 | sampling_file_init(lqs, njf); |
| 591 | |
| 592 | ND_SD_JOURNAL_STATUS tmp_status = nd_sd_journal_query_one_file(filename, wb, facets, njf, lqs); |
| 593 | |
| 594 | rows_useful = lqs->c.rows_useful - rows_useful; |
| 595 | rows_read = lqs->c.rows_read - rows_read; |
| 596 | bytes_read = lqs->c.bytes_read - bytes_read; |
| 597 | matches_setup_ut = lqs->c.matches_setup_ut - matches_setup_ut; |
| 598 | fs_calls = fstat_thread_calls - fs_calls; |
| 599 | fs_cached = fstat_thread_cached_responses - fs_cached; |
| 600 | |
| 601 | ended_ut = now_monotonic_usec(); |
| 602 | duration_ut = ended_ut - started_ut; |
| 603 | |
| 604 | if (duration_ut > max_duration_ut) |
| 605 | max_duration_ut = duration_ut; |
| 606 | |
| 607 | progress_duration_ut += duration_ut; |
| 608 | if (progress_duration_ut >= ND_SD_JOURNAL_PROGRESS_EVERY_UT) { |
| 609 | progress_duration_ut = 0; |
| 610 | netdata_mutex_lock(&stdout_mutex); |
| 611 | pluginsd_function_progress_to_stdout(lqs->rq.transaction, f + 1, files_used); |
| 612 | netdata_mutex_unlock(&stdout_mutex); |
| 613 | } |
| 614 | |
| 615 | buffer_json_add_array_item_object(wb); // journal file |
| 616 | { |
| 617 | // information about the file |
| 618 | buffer_json_member_add_string(wb, "_filename", filename); |
| 619 | buffer_json_member_add_uint64(wb, "_source_type", njf->source_type); |
| 620 | buffer_json_member_add_string(wb, "_source", string2str(njf->source)); |
| 621 | buffer_json_member_add_uint64(wb, "_last_modified_ut", njf->file_last_modified_ut); |
| 622 | buffer_json_member_add_uint64(wb, "_msg_first_ut", njf->msg_first_ut); |
| 623 | buffer_json_member_add_uint64(wb, "_msg_last_ut", njf->msg_last_ut); |
| 624 | buffer_json_member_add_uint64(wb, "_journal_vs_realtime_delta_ut", njf->max_journal_vs_realtime_delta_ut); |
| 625 | |
| 626 | // information about the current use of the file |
| 627 | buffer_json_member_add_uint64(wb, "duration_ut", ended_ut - started_ut); |
| 628 | buffer_json_member_add_uint64(wb, "rows_read", rows_read); |
| 629 | buffer_json_member_add_uint64(wb, "rows_useful", rows_useful); |
| 630 | buffer_json_member_add_double( |
| 631 | wb, "rows_per_second", (double)rows_read / (double)duration_ut * (double)USEC_PER_SEC); |
| 632 | buffer_json_member_add_uint64(wb, "bytes_read", bytes_read); |
| 633 | buffer_json_member_add_double( |
| 634 | wb, "bytes_per_second", (double)bytes_read / (double)duration_ut * (double)USEC_PER_SEC); |
| 635 | buffer_json_member_add_uint64(wb, "duration_matches_ut", matches_setup_ut); |
| 636 | buffer_json_member_add_uint64(wb, "fstat_query_calls", fs_calls); |
| 637 | buffer_json_member_add_uint64(wb, "fstat_query_cached_responses", fs_cached); |
| 638 | |
| 639 | if (lqs->rq.sampling) { |
| 640 | buffer_json_member_add_object(wb, "_sampling"); |
| 641 | { |
| 642 | buffer_json_member_add_uint64(wb, "sampled", lqs->c.samples_per_file.sampled); |
| 643 | buffer_json_member_add_uint64(wb, "unsampled", lqs->c.samples_per_file.unsampled); |
| 644 | buffer_json_member_add_uint64(wb, "estimated", lqs->c.samples_per_file.estimated); |
| 645 | } |
| 646 | buffer_json_object_close(wb); // _sampling |
| 647 | } |
| 648 | } |
| 649 | buffer_json_object_close(wb); // journal file |
| 650 | |
| 651 | bool stop = false; |
| 652 | switch (tmp_status) { |
| 653 | case ND_SD_JOURNAL_OK: |
| 654 | case ND_SD_JOURNAL_NO_FILE_MATCHED: |
| 655 | status = (status == ND_SD_JOURNAL_OK) ? ND_SD_JOURNAL_OK : tmp_status; |
| 656 | break; |
| 657 | |
| 658 | case ND_SD_JOURNAL_FAILED_TO_OPEN: |
| 659 | case ND_SD_JOURNAL_FAILED_TO_SEEK: |
| 660 | partial = true; |
| 661 | if (status == ND_SD_JOURNAL_NO_FILE_MATCHED) |
| 662 | status = tmp_status; |
| 663 | break; |
| 664 | |
| 665 | case ND_SD_JOURNAL_CANCELLED: |
| 666 | case ND_SD_JOURNAL_TIMED_OUT: |
| 667 | partial = true; |
| 668 | stop = true; |
| 669 | status = tmp_status; |
| 670 | break; |
| 671 | |
| 672 | case ND_SD_JOURNAL_NOT_MODIFIED: |
| 673 | internal_fatal(true, "this should never be returned here"); |
| 674 | break; |
| 675 | } |
| 676 | |
| 677 | if (stop) |
| 678 | break; |
| 679 | } |
| 680 | buffer_json_array_close(wb); // _journal_files |
| 681 | |
| 682 | // release the files |
| 683 | for (size_t f = 0; f < files_used; f++) |
| 684 | dictionary_acquired_item_release(nd_journal_files_registry, file_items[f]); |
| 685 | |
| 686 | switch (status) { |
| 687 | case ND_SD_JOURNAL_OK: |
| 688 | if (lqs->rq.if_modified_since && !lqs->c.rows_useful) |
| 689 | return rrd_call_function_error( |
| 690 | wb, "No additional useful data since the previous call.", HTTP_RESP_NOT_MODIFIED); |
| 691 | break; |
| 692 | |
| 693 | case ND_SD_JOURNAL_TIMED_OUT: |
| 694 | case ND_SD_JOURNAL_NO_FILE_MATCHED: |
| 695 | break; |
| 696 | |
| 697 | case ND_SD_JOURNAL_CANCELLED: |
| 698 | return rrd_call_function_error(wb, "Request cancelled.", HTTP_RESP_CLIENT_CLOSED_REQUEST); |
| 699 | |
| 700 | case ND_SD_JOURNAL_NOT_MODIFIED: |
| 701 | return rrd_call_function_error(wb, "No new data since the previous call.", HTTP_RESP_NOT_MODIFIED); |
| 702 | |
| 703 | case ND_SD_JOURNAL_FAILED_TO_OPEN: |
| 704 | return rrd_call_function_error(wb, "Failed to open systemd journal file.", HTTP_RESP_INTERNAL_SERVER_ERROR); |
| 705 | |
| 706 | case ND_SD_JOURNAL_FAILED_TO_SEEK: |
| 707 | return rrd_call_function_error( |
| 708 | wb, "Failed to seek in systemd journal file.", HTTP_RESP_INTERNAL_SERVER_ERROR); |
| 709 | |
| 710 | default: |
| 711 | return rrd_call_function_error(wb, "Unknown status", HTTP_RESP_INTERNAL_SERVER_ERROR); |
| 712 | } |
| 713 | |
| 714 | buffer_json_member_add_uint64(wb, "status", HTTP_RESP_OK); |
| 715 | buffer_json_member_add_boolean(wb, "partial", partial); |
| 716 | buffer_json_member_add_string(wb, "type", "table"); |
| 717 | |
| 718 | // build a message for the query |
| 719 | if (!lqs->rq.data_only) { |
| 720 | CLEAN_BUFFER *msg = buffer_create(0, NULL); |
| 721 | CLEAN_BUFFER *msg_description = buffer_create(0, NULL); |
| 722 | ND_LOG_FIELD_PRIORITY msg_priority = NDLP_INFO; |
| 723 | |
| 724 | if (!nd_journal_files_completed_once()) { |
| 725 | buffer_strcat(msg, "Journals are still being scanned. "); |
| 726 | buffer_strcat( |
| 727 | msg_description, |
| 728 | "LIBRARY SCAN: The journal files are still being scanned, you are probably viewing incomplete data. "); |
| 729 | msg_priority = NDLP_WARNING; |
| 730 | } |
| 731 | |
| 732 | if (partial) { |
| 733 | buffer_strcat(msg, "Query timed-out, incomplete data. "); |
| 734 | buffer_strcat( |
| 735 | msg_description, |
| 736 | "QUERY TIMEOUT: The query timed out and may not include all the data of the selected window. "); |
| 737 | msg_priority = NDLP_WARNING; |
| 738 | } |
| 739 | |
| 740 | if (lqs->c.samples.estimated || lqs->c.samples.unsampled) { |
| 741 | double percent = (double)(lqs->c.samples.sampled * 100.0 / |
| 742 | (lqs->c.samples.estimated + lqs->c.samples.unsampled + lqs->c.samples.sampled)); |
| 743 | buffer_sprintf(msg, "%.2f%% real data", percent); |
| 744 | buffer_sprintf(msg_description, "ACTUAL DATA: The filters counters reflect %0.2f%% of the data. ", percent); |
| 745 | msg_priority = MIN(msg_priority, NDLP_NOTICE); |
| 746 | } |
| 747 | |
| 748 | if (lqs->c.samples.unsampled) { |
| 749 | double percent = (double)(lqs->c.samples.unsampled * 100.0 / |
| 750 | (lqs->c.samples.estimated + lqs->c.samples.unsampled + lqs->c.samples.sampled)); |
| 751 | buffer_sprintf(msg, ", %.2f%% unsampled", percent); |
| 752 | buffer_sprintf( |
| 753 | msg_description, |
| 754 | "UNSAMPLED DATA: %0.2f%% of the events exist and have been counted, but their values have not been evaluated, so they are not included in the filters counters. ", |
| 755 | percent); |
| 756 | msg_priority = MIN(msg_priority, NDLP_NOTICE); |
| 757 | } |
| 758 | |
| 759 | if (lqs->c.samples.estimated) { |
| 760 | double percent = (double)(lqs->c.samples.estimated * 100.0 / |
| 761 | (lqs->c.samples.estimated + lqs->c.samples.unsampled + lqs->c.samples.sampled)); |
| 762 | buffer_sprintf(msg, ", %.2f%% estimated", percent); |
| 763 | buffer_sprintf( |
| 764 | msg_description, |
| 765 | "ESTIMATED DATA: The query selected a large amount of data, so to avoid delaying too much, the presented data are estimated by %0.2f%%. ", |
| 766 | percent); |
| 767 | msg_priority = MIN(msg_priority, NDLP_NOTICE); |
| 768 | } |
| 769 | |
| 770 | buffer_json_member_add_object(wb, "message"); |
| 771 | if (buffer_tostring(msg)) { |
| 772 | buffer_json_member_add_string(wb, "title", buffer_tostring(msg)); |
| 773 | buffer_json_member_add_string(wb, "description", buffer_tostring(msg_description)); |
| 774 | buffer_json_member_add_string(wb, "status", nd_log_id2priority(msg_priority)); |
| 775 | } |
| 776 | // else send an empty object if there is nothing to tell |
| 777 | buffer_json_object_close(wb); // message |
| 778 | } |
| 779 | |
| 780 | if (!lqs->rq.data_only) { |
| 781 | buffer_json_member_add_time_t(wb, "update_every", 1); |
| 782 | buffer_json_member_add_string(wb, "help", ND_SD_JOURNAL_FUNCTION_DESCRIPTION); |
| 783 | } |
| 784 | |
| 785 | if (!lqs->rq.data_only || lqs->rq.tail) |
| 786 | buffer_json_member_add_uint64(wb, "last_modified", lqs->last_modified); |
| 787 | |
| 788 | facets_sort_and_reorder_keys(facets); |
| 789 | facets_report(facets, wb, used_hashes_registry); |
| 790 | |
| 791 | wb->expires = now_realtime_sec() + (lqs->rq.data_only ? 3600 : 0); |
| 792 | buffer_json_member_add_time_t(wb, "expires", wb->expires); |
| 793 | |
| 794 | buffer_json_member_add_object(wb, "_fstat_caching"); |
| 795 | { |
| 796 | buffer_json_member_add_uint64(wb, "calls", fstat_thread_calls); |
| 797 | buffer_json_member_add_uint64(wb, "cached", fstat_thread_cached_responses); |
| 798 | } |
| 799 | buffer_json_object_close(wb); // _fstat_caching |
| 800 | |
| 801 | if (lqs->rq.sampling) { |
| 802 | buffer_json_member_add_object(wb, "_sampling"); |
| 803 | { |
| 804 | buffer_json_member_add_uint64(wb, "sampled", lqs->c.samples.sampled); |
| 805 | buffer_json_member_add_uint64(wb, "unsampled", lqs->c.samples.unsampled); |
| 806 | buffer_json_member_add_uint64(wb, "estimated", lqs->c.samples.estimated); |
| 807 | } |
| 808 | buffer_json_object_close(wb); // _sampling |
| 809 | } |
| 810 | |
| 811 | wb->content_type = CT_APPLICATION_JSON; |
| 812 | wb->response_code = HTTP_RESP_OK; |
| 813 | return wb->response_code; |
| 814 | } |
| 815 | |
| 816 | #endif //NETDATA_SYSTEMD_JOURNAL_EXECUTE_H |