1use crate::explorer::{
2 ExplorerSamplingState, empty_histogram_for_query, histogram_bucket_count_for_query,
3};
4use crate::{
5 Direction, ExplorerAnchor, ExplorerControl, ExplorerFieldMode, ExplorerFilter,
6 ExplorerFtsPattern, ExplorerHistogram, ExplorerProgress, ExplorerQuery, ExplorerResult,
7 ExplorerRow, ExplorerSampling, ExplorerStats, ExplorerStopReason, ExplorerStrategy, FileHeader,
8 FileReader, ReaderOptions, Result, SdkError,
9};
10use chrono::{DateTime, Utc};
11use serde_json::{Map, Value, json};
12use std::cell::RefCell;
13use std::cmp::{Ordering, Reverse};
14use std::collections::{BTreeMap, BTreeSet, BinaryHeap, HashSet, VecDeque};
15#[cfg(unix)]
16use std::ffi::CStr;
17use std::path::{Path, PathBuf};
18use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
19
20const DEFAULT_FUNCTION_NAME: &str = "systemd-journal";
21const DEFAULT_SOURCE_SELECTOR_NAME: &str = "Journal Sources";
22const DEFAULT_SOURCE_SELECTOR_HELP: &str = "Select the logs source to query";
23const DEFAULT_ITEMS_TO_RETURN: usize = 200;
24const DEFAULT_TIME_WINDOW_SECONDS: i64 = 3600;
25const DEFAULT_ITEMS_SAMPLING: u64 = 1_000_000;
26const DATA_ONLY_CHECK_EVERY_ROWS: u64 = 128;
27const API_RELATIVE_TIME_MAX_SECONDS: i64 = 3 * 365 * 86_400;
28const NETDATA_MISSING_AFTER_RELATIVE_SECONDS: i64 = 600;
29const DEFAULT_HISTOGRAM_BUCKETS: usize = 150;
30const EFFECTIVELY_DISABLED_TIMEOUT_SECONDS: u64 = 100 * 365 * 24 * 60 * 60;
31const NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC: u64 = 5_000_000;
32const NETDATA_JOURNAL_VS_REALTIME_DELTA_MAX_USEC: u64 = 2 * 60 * 1_000_000;
33const NETDATA_EMPTY_STRING_FACET_HASH_ID: &str = "CzGfAU2z3TC";
34const NETDATA_UNAVAILABLE_FIELD_LABEL: &str = "[unavailable field]";
35const NETDATA_FACET_MAX_VALUE_LENGTH: usize = 8192;
36const NETDATA_MAX_DIRECTORY_SCAN_DEPTH: usize = 64;
37const NETDATA_MAX_DIRECTORY_SCAN_COUNT: usize = 8192;
38const SOURCE_TYPE_ALL: u64 = 1 << 0;
39const SOURCE_TYPE_LOCAL_ALL: u64 = 1 << 1;
40const SOURCE_TYPE_REMOTE_ALL: u64 = 1 << 2;
41const SOURCE_TYPE_LOCAL_SYSTEM: u64 = 1 << 3;
42const SOURCE_TYPE_LOCAL_USER: u64 = 1 << 4;
43const SOURCE_TYPE_LOCAL_NAMESPACE: u64 = 1 << 5;
44const SOURCE_TYPE_LOCAL_OTHER: u64 = 1 << 6;
45
46pub const NETDATA_SOURCE_TYPE_ALL: u64 = SOURCE_TYPE_ALL;
47pub const NETDATA_SOURCE_TYPE_LOCAL_ALL: u64 = SOURCE_TYPE_LOCAL_ALL;
48pub const NETDATA_SOURCE_TYPE_REMOTE_ALL: u64 = SOURCE_TYPE_REMOTE_ALL;
49pub const NETDATA_SOURCE_TYPE_LOCAL_SYSTEM: u64 = SOURCE_TYPE_LOCAL_SYSTEM;
50pub const NETDATA_SOURCE_TYPE_LOCAL_USER: u64 = SOURCE_TYPE_LOCAL_USER;
51pub const NETDATA_SOURCE_TYPE_LOCAL_NAMESPACE: u64 = SOURCE_TYPE_LOCAL_NAMESPACE;
52pub const NETDATA_SOURCE_TYPE_LOCAL_OTHER: u64 = SOURCE_TYPE_LOCAL_OTHER;
53
54const NETDATA_ACCEPTED_PARAMS: &[&str] = &[
55 "info",
56 "__logs_sources",
57 "after",
58 "before",
59 "anchor",
60 "direction",
61 "last",
62 "query",
63 "facets",
64 "histogram",
65 "if_modified_since",
66 "data_only",
67 "delta",
68 "tail",
69 "sampling",
70 "slice",
71];
72
73const SYSTEMD_DEFAULT_VIEW_KEYS: &[&str] = &[
74 "_HOSTNAME",
75 "ND_JOURNAL_PROCESS",
76 "MESSAGE",
77 "PRIORITY",
78 "SYSLOG_FACILITY",
79 "ERRNO",
80 "ND_JOURNAL_FILE",
81 "SYSLOG_IDENTIFIER",
82 "UNIT",
83 "USER_UNIT",
84 "MESSAGE_ID",
85 "_BOOT_ID",
86 "_SYSTEMD_OWNER_UID",
87 "_UID",
88 "OBJECT_SYSTEMD_OWNER_UID",
89 "OBJECT_UID",
90 "_GID",
91 "OBJECT_GID",
92 "_CAP_EFFECTIVE",
93 "_AUDIT_LOGINUID",
94 "OBJECT_AUDIT_LOGINUID",
95 "_SOURCE_REALTIME_TIMESTAMP",
96];
97
98const SYSTEMD_DEFAULT_FACETS: &[&str] = &[
99 "_HOSTNAME",
100 "PRIORITY",
101 "SYSLOG_FACILITY",
102 "ERRNO",
103 "SYSLOG_IDENTIFIER",
104 "UNIT",
105 "USER_UNIT",
106 "MESSAGE_ID",
107 "_BOOT_ID",
108 "_SYSTEMD_OWNER_UID",
109 "_UID",
110 "OBJECT_SYSTEMD_OWNER_UID",
111 "OBJECT_UID",
112 "_GID",
113 "OBJECT_GID",
114 "_AUDIT_LOGINUID",
115 "OBJECT_AUDIT_LOGINUID",
116 "CODE_FILE",
117 "_SYSTEMD_UNIT",
118 "_SYSTEMD_USER_SLICE",
119 "CODE_FUNC",
120 "_TRANSPORT",
121 "_COMM",
122 "_RUNTIME_SCOPE",
123 "_MACHINE_ID",
124 "_SYSTEMD_SLICE",
125 "UNIT_RESULT",
126 "_SYSTEMD_CGROUP",
127 "_EXE",
128 "_SYSTEMD_USER_UNIT",
129 "_SYSTEMD_SESSION",
130 "COREDUMP_CGROUP",
131 "COREDUMP_USER_UNIT",
132 "COREDUMP_UNIT",
133 "COREDUMP_SIGNAL_NAME",
134 "COREDUMP_COMM",
135 "_UDEV_DEVNODE",
136 "_KERNEL_SUBSYSTEM",
137 "OBJECT_EXE",
138 "OBJECT_SYSTEMD_CGROUP",
139 "OBJECT_COMM",
140 "OBJECT_SYSTEMD_UNIT",
141 "OBJECT_SYSTEMD_USER_UNIT",
142 "_SELINUX_CONTEXT",
143 "_NAMESPACE",
144 "OBJECT_SYSTEMD_SESSION",
145 "CONTAINER_ID",
146 "CONTAINER_NAME",
147 "CONTAINER_TAG",
148 "IMAGE_NAME",
149 "ND_NIDL_NODE",
150 "ND_NIDL_CONTEXT",
151 "ND_LOG_SOURCE",
152 "ND_ALERT_NAME",
153 "ND_ALERT_CLASS",
154 "ND_ALERT_COMPONENT",
155 "ND_ALERT_TYPE",
156 "ND_ALERT_STATUS",
157];
158
159#[derive(Debug, Clone)]
160pub struct NetdataFunctionConfig {
161 pub function_name: String,
162 pub source_selector_name: String,
163 pub source_selector_help: String,
164 pub default_facets: Vec<String>,
165 pub default_view_keys: Vec<String>,
166 pub default_histogram: Option<String>,
167 pub reader_options: ReaderOptions,
168 pub explorer_strategy: ExplorerStrategy,
169}
170
171impl NetdataFunctionConfig {
172 pub fn systemd_journal() -> Self {
173 Self {
174 function_name: DEFAULT_FUNCTION_NAME.to_string(),
175 source_selector_name: DEFAULT_SOURCE_SELECTOR_NAME.to_string(),
176 source_selector_help: DEFAULT_SOURCE_SELECTOR_HELP.to_string(),
177 default_facets: SYSTEMD_DEFAULT_FACETS
178 .iter()
179 .map(|field| (*field).to_string())
180 .collect(),
181 default_view_keys: SYSTEMD_DEFAULT_VIEW_KEYS
182 .iter()
183 .map(|field| (*field).to_string())
184 .collect(),
185 default_histogram: Some("PRIORITY".to_string()),
186 reader_options: ReaderOptions::snapshot(),
187 explorer_strategy: ExplorerStrategy::Traversal,
188 }
189 }
190}
191
192impl Default for NetdataFunctionConfig {
193 fn default() -> Self {
194 Self::systemd_journal()
195 }
196}
197
198#[derive(Debug, Default)]
199pub struct DisplayContext {
200 boot_first_realtime: BTreeMap<Vec<u8>, u64>,
201 uid_display_cache: RefCell<BTreeMap<String, String>>,
202 gid_display_cache: RefCell<BTreeMap<String, String>>,
203}
204
205#[derive(Debug, Clone, Copy)]
206pub enum DisplayScope {
207 Data,
208 Facet,
209 Histogram,
210}
211
212pub trait NetdataFunctionProfile {
213 fn field_display_value(
214 &self,
215 _context: &DisplayContext,
216 _scope: DisplayScope,
217 _field: &str,
218 value: &[u8],
219 ) -> Value {
220 Value::String(String::from_utf8_lossy(value).into_owned())
221 }
222
223 fn facet_option_name(&self, context: &DisplayContext, field: &str, raw_value: &[u8]) -> String {
224 match self.field_display_value(context, DisplayScope::Facet, field, raw_value) {
225 Value::String(value) => value,
226 other => other.to_string(),
227 }
228 }
229
230 fn row_options(&self, fields: &BTreeMap<String, Vec<Vec<u8>>>) -> Value {
231 if let Some(priority) = first_value(fields, "PRIORITY") {
232 return json!({ "severity": priority_to_row_severity(priority) });
233 }
234 json!({ "severity": "normal" })
235 }
236}
237
238#[derive(Debug, Clone, Copy, Default)]
239pub struct SystemdJournalProfile;
240
241#[derive(Debug, Clone, Copy, Default)]
242pub struct SystemdJournalPluginProfile;
243
244impl NetdataFunctionProfile for SystemdJournalProfile {
245 fn field_display_value(
246 &self,
247 context: &DisplayContext,
248 scope: DisplayScope,
249 field: &str,
250 value: &[u8],
251 ) -> Value {
252 systemd_field_display_value(context, scope, field, value, false)
253 }
254}
255
256impl NetdataFunctionProfile for SystemdJournalPluginProfile {
257 fn field_display_value(
258 &self,
259 context: &DisplayContext,
260 scope: DisplayScope,
261 field: &str,
262 value: &[u8],
263 ) -> Value {
264 systemd_field_display_value(context, scope, field, value, true)
265 }
266}
267
268#[derive(Debug, Clone)]
269pub struct NetdataJournalFunction<P = SystemdJournalProfile> {
270 config: NetdataFunctionConfig,
271 profile: P,
272}
273
274#[derive(Debug, Clone)]
275pub struct NetdataFunctionProgress {
276 pub current_file: usize,
277 pub total_files: usize,
278 pub matched_files: u64,
279 pub skipped_files: u64,
280 pub stats: ExplorerStats,
281 pub elapsed: Duration,
282}
283
284#[derive(Debug, Clone, Default, PartialEq, Eq)]
285pub struct NetdataJournalFileMetadata {
286 pub source_type: Option<u64>,
287 pub source_name: Option<String>,
288 pub file_last_modified_usec: Option<u64>,
289 pub msg_first_realtime_usec: Option<u64>,
290 pub msg_last_realtime_usec: Option<u64>,
291 pub journal_vs_realtime_delta_usec: Option<u64>,
292}
293
294pub trait NetdataFunctionState {
295 fn file_metadata(&self, _path: &Path) -> Option<NetdataJournalFileMetadata> {
296 None
297 }
298
299 fn update_file_journal_vs_realtime_delta_usec(&mut self, _path: &Path, _delta_usec: u64) {}
300}
301
302pub struct NetdataFunctionRunOptions<'a> {
303 pub timeout: Option<Duration>,
304 pub progress_callback: Option<&'a mut dyn FnMut(NetdataFunctionProgress)>,
305 pub cancellation_callback: Option<&'a dyn Fn() -> bool>,
306 pub state: Option<&'a mut dyn NetdataFunctionState>,
307 pub progress_interval: Duration,
308}
309
310impl NetdataFunctionRunOptions<'_> {
311 pub fn from_timeout_seconds(seconds: u64) -> Self {
312 let seconds = if seconds == 0 {
313 EFFECTIVELY_DISABLED_TIMEOUT_SECONDS
314 } else {
315 seconds
316 };
317 Self {
318 timeout: Some(Duration::from_secs(seconds)),
319 progress_callback: None,
320 cancellation_callback: None,
321 state: None,
322 progress_interval: Duration::from_millis(250),
323 }
324 }
325}
326
327impl Default for NetdataFunctionRunOptions<'_> {
328 fn default() -> Self {
329 Self {
330 timeout: Some(Duration::from_secs(EFFECTIVELY_DISABLED_TIMEOUT_SECONDS)),
331 progress_callback: None,
332 cancellation_callback: None,
333 state: None,
334 progress_interval: Duration::from_millis(250),
335 }
336 }
337}
338
339impl NetdataJournalFunction<SystemdJournalProfile> {
340 pub fn systemd_journal() -> Self {
341 Self {
342 config: NetdataFunctionConfig::systemd_journal(),
343 profile: SystemdJournalProfile,
344 }
345 }
346}
347
348impl NetdataJournalFunction<SystemdJournalPluginProfile> {
349 pub fn systemd_journal_plugin_compatible() -> Self {
350 Self {
351 config: NetdataFunctionConfig::systemd_journal(),
352 profile: SystemdJournalPluginProfile,
353 }
354 }
355}
356
357impl<P> NetdataJournalFunction<P>
358where
359 P: NetdataFunctionProfile,
360{
361 pub fn new(mut config: NetdataFunctionConfig, profile: P) -> Self {
362 if config.source_selector_name.is_empty() {
363 config.source_selector_name = DEFAULT_SOURCE_SELECTOR_NAME.to_string();
364 }
365 if config.source_selector_help.is_empty() {
366 config.source_selector_help = DEFAULT_SOURCE_SELECTOR_HELP.to_string();
367 }
368 Self { config, profile }
369 }
370
371 pub fn run_directory_request_json(&self, directory: &Path, request: &Value) -> Result<Value> {
372 self.run_directory_request_json_with_options(
373 directory,
374 request,
375 NetdataFunctionRunOptions::default(),
376 )
377 }
378
379 pub fn run_directory_request_json_with_options(
380 &self,
381 directory: &Path,
382 request: &Value,
383 mut options: NetdataFunctionRunOptions<'_>,
384 ) -> Result<Value> {
385 let request = NetdataRequest::parse(request, &self.config)?;
386 let collection = collect_journal_files(directory)?;
387 let paths = collection.files;
388 if request.info {
389 return Ok(self.info_response(request.echo, &paths, &options));
390 }
391 let annotation_paths = paths.clone();
392
393 let selected =
394 select_journal_files_for_request(paths, &request, self.config.reader_options, &options);
395 if let Some(response) = not_modified_before_scan_response(&request, &selected) {
396 return Ok(response);
397 }
398 let selected_files = selected.files;
399 let deadline = options.timeout.map(|timeout| Instant::now() + timeout);
400 let mut combined = self.explore_files(&selected_files, &request, deadline, &mut options)?;
401 self.finalize_combined_result(
402 &request,
403 &selected_files,
404 deadline,
405 &mut options,
406 &mut combined,
407 collection.skipped,
408 collection.errors,
409 )?;
410 Ok(self.query_response(request, &annotation_paths, combined))
411 }
412
413 fn finalize_combined_result(
414 &self,
415 request: &NetdataRequest,
416 selected_files: &[SelectedJournalFile],
417 deadline: Option<Instant>,
418 options: &mut NetdataFunctionRunOptions<'_>,
419 combined: &mut CombinedResult,
420 skipped_files: u64,
421 file_errors: Vec<String>,
422 ) -> Result<()> {
423 combined.skipped_files = combined.skipped_files.saturating_add(skipped_files);
424 combined.file_errors.extend(file_errors);
425 if combined.cancelled {
426 return Ok(());
427 }
428 if !request.data_only {
429 combined.add_zero_count_facet_values_from_files(
430 &request.facets,
431 self.config.reader_options,
432 );
433 combined.add_zero_count_selected_filter_values(request);
434 }
435 if should_collect_unfiltered_facet_vocabulary(request, combined) {
436 let vocabulary = self.explore_files(
437 selected_files,
438 &request.unfiltered_vocabulary(),
439 deadline,
440 options,
441 )?;
442 combined.add_zero_count_facet_values(&vocabulary.facets);
443 }
444 Ok(())
445 }
446
447 pub fn run_directory_request_bytes(&self, directory: &Path, request: &[u8]) -> Result<Value> {
448 self.run_directory_request_bytes_with_options(
449 directory,
450 request,
451 NetdataFunctionRunOptions::default(),
452 )
453 }
454
455 pub fn run_directory_request_bytes_with_options(
456 &self,
457 directory: &Path,
458 request: &[u8],
459 options: NetdataFunctionRunOptions<'_>,
460 ) -> Result<Value> {
461 let request: Value = serde_json::from_slice(request).map_err(|err| {
462 SdkError::InvalidPath(format!("invalid Netdata function JSON: {err}"))
463 })?;
464 self.run_directory_request_json_with_options(directory, &request, options)
465 }
466
467 fn explore_files(
468 &self,
469 files: &[SelectedJournalFile],
470 request: &NetdataRequest,
471 deadline: Option<Instant>,
472 options: &mut NetdataFunctionRunOptions<'_>,
473 ) -> Result<CombinedResult> {
474 let query = request.to_explorer_query(
475 files.len() as u64,
476 None,
477 NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC,
478 );
479 let mut combined = CombinedResult::default();
480 let page_window = RefCell::new(NetdataPageWindow::for_request(request));
481 combined.sampling_enabled = query.sampling.is_some();
482 let mut sampling_state =
483 ExplorerSamplingState::for_query(&query, histogram_bucket_count_for_query(&query));
484 let started = Instant::now();
485 let total_files = files.len();
486 for (file_index, file) in files.iter().enumerate() {
487 let path = &file.path;
488 if should_stop_before_file(&mut combined, deadline, options) {
489 break;
490 }
491 let Some(mut reader) = self.open_file_for_explore(
492 path,
493 &mut combined,
494 options,
495 progress_context(file_index, total_files, started),
496 )?
497 else {
498 continue;
499 };
500 combined.matched_files = combined.matched_files.saturating_add(1);
501 combined.matched_paths.push(path.clone());
502 let query = request.file_query(files.len(), reader.header(), &file.order);
503 collect_column_fields_for_file(&mut reader, request, path, &mut combined);
504 let explored = self.explore_single_file(
505 &mut reader,
506 request,
507 &query,
508 deadline,
509 options,
510 &combined,
511 &page_window,
512 sampling_state.as_mut(),
513 progress_context(file_index, total_files, started),
514 file.order.journal_vs_realtime_delta_usec,
515 );
516 let Some((result, stop_reason)) = record_explore_result(explored, path, &mut combined)
517 else {
518 continue;
519 };
520 if finish_explored_file(
521 options,
522 request,
523 file,
524 &query,
525 result,
526 stop_reason,
527 &mut combined,
528 files,
529 file_index,
530 progress_context(file_index, total_files, started),
531 )? {
532 break;
533 }
534 }
535 combined.expand_row_payloads(self.config.reader_options);
536 combined.page_counters = Some(page_window.into_inner().counters());
537 Ok(combined)
538 }
539
540 fn open_file_for_explore(
541 &self,
542 path: &Path,
543 combined: &mut CombinedResult,
544 options: &mut NetdataFunctionRunOptions<'_>,
545 progress: ProgressContext,
546 ) -> Result<Option<FileReader>> {
547 match FileReader::open_with_options(path, self.config.reader_options) {
548 Ok(reader) => Ok(Some(reader)),
549 Err(err) => {
550 combined.skipped_files = combined.skipped_files.saturating_add(1);
551 combined
552 .file_errors
553 .push(format!("{}: {err}", path.display()));
554 emit_progress_for_combined(options, combined, progress);
555 Ok(None)
556 }
557 }
558 }
559
560 #[allow(clippy::too_many_arguments)]
561 fn explore_single_file(
562 &self,
563 reader: &mut FileReader,
564 request: &NetdataRequest,
565 query: &ExplorerQuery,
566 deadline: Option<Instant>,
567 options: &mut NetdataFunctionRunOptions<'_>,
568 combined: &CombinedResult,
569 page_window: &RefCell<NetdataPageWindow>,
570 sampling_state: Option<&mut ExplorerSamplingState>,
571 progress: ProgressContext,
572 realtime_delta_usec: u64,
573 ) -> Result<(ExplorerResult, Option<ExplorerStopReason>)> {
574 let cancellation_callback = options.cancellation_callback;
575 let progress_interval = options.progress_interval;
576 let mut explorer_progress = |explorer_progress: ExplorerProgress| {
577 emit_explorer_progress(options, combined, explorer_progress, progress);
578 };
579 let mut control = ExplorerControl::new();
580 control.set_deadline(deadline);
581 control.set_cancellation_callback(cancellation_callback);
582 control.set_progress_interval(progress_interval);
583 control.set_progress_callback(Some(&mut explorer_progress));
584 control.set_sampling_state(sampling_state);
585 let mut candidate_row =
586 |realtime_usec| page_window.borrow().candidate_to_keep(realtime_usec);
587 control.set_candidate_row_callback(Some(&mut candidate_row));
588 let mut matched_row = |realtime_usec, rows_matched| {
589 delta_scan_can_stop(
590 request,
591 page_window,
592 realtime_usec,
593 rows_matched,
594 realtime_delta_usec,
595 )
596 };
597 control.set_matched_row_callback(Some(&mut matched_row));
598 let result = reader.explore_with_strategy_cursor_rows_controlled(
599 query,
600 self.config.explorer_strategy,
601 &mut control,
602 )?;
603 Ok((result, control.stop_reason()))
604 }
605
606 fn info_response(
607 &self,
608 echo: Value,
609 paths: &[PathBuf],
610 options: &NetdataFunctionRunOptions<'_>,
611 ) -> Value {
612 json!({
613 "_request": echo,
614 "versions": { "netdata_function_api": 1, "sdk": env!("CARGO_PKG_VERSION") },
615 "v": 3,
616 "accepted_params": self.accepted_params_from_fields(&[]),
617 "required_params": self.required_source_params(paths, options),
618 "show_ids": true,
619 "has_history": true,
620 "pagination": {
621 "enabled": true,
622 "key": "anchor",
623 "column": "timestamp",
624 "units": "timestamp_usec",
625 },
626 "status": 200,
627 "type": "table",
628 "help": "Netdata-compatible journal log function backed by the systemd journal SDK"
629 })
630 }
631
632 fn query_response(
633 &self,
634 request: NetdataRequest,
635 annotation_paths: &[PathBuf],
636 combined: CombinedResult,
637 ) -> Value {
638 if combined.cancelled {
639 return netdata_function_error(499, "Request cancelled.");
640 }
641 let artifacts = self.query_response_artifacts(&request, annotation_paths, &combined);
642 let mut response = base_query_response(&request, &combined, &artifacts);
643 let Some(object) = response.as_object_mut() else {
644 return netdata_function_error(500, "Internal Netdata function response error.");
645 };
646 self.add_query_response_metadata(object, &request, &combined, artifacts);
647 response
648 }
649
650 fn query_response_artifacts(
651 &self,
652 request: &NetdataRequest,
653 annotation_paths: &[PathBuf],
654 combined: &CombinedResult,
655 ) -> QueryResponseArtifacts {
656 let reportable_facet_fields = combined.reportable_facet_fields_bytes(&request.facets);
657 let reportable_facet_field_names = string_fields(&reportable_facet_fields);
658 let columns = self.build_columns(
659 &request,
660 &reportable_facet_fields,
661 &combined.rows,
662 &combined.facets,
663 &combined.column_fields,
664 );
665 let boot_ids = response_boot_ids(
666 &columns.order,
667 &combined.rows,
668 &combined.facets,
669 combined.histogram.as_ref(),
670 );
671 let context = DisplayContext {
672 boot_first_realtime: collect_boot_first_realtime(
673 annotation_paths,
674 self.config.reader_options,
675 &boot_ids,
676 ),
677 ..DisplayContext::default()
678 };
679 let data =
680 self.build_data_rows(&context, &columns.order, &combined.rows, request.direction);
681 let facets = self.build_facets(&context, &reportable_facet_fields, &combined.facets);
682 let histogram = self.query_histogram_artifact(request, combined, &context);
683 let message = query_message(combined.timed_out, &combined.stats);
684 let items = response_items(request, combined, data.len() as u64);
685 QueryResponseArtifacts {
686 reportable_facet_field_names,
687 columns,
688 data,
689 facets,
690 histogram,
691 message,
692 items,
693 }
694 }
695
696 fn add_query_response_metadata(
697 &self,
698 object: &mut Map<String, Value>,
699 request: &NetdataRequest,
700 combined: &CombinedResult,
701 artifacts: QueryResponseArtifacts,
702 ) {
703 if !request.data_only {
704 self.add_full_query_response_metadata(object, request, combined, &artifacts);
705 } else if request.histogram.is_some() {
706 object.insert(
707 "available_histograms".to_string(),
708 self.available_histograms(request, combined),
709 );
710 }
711 add_last_modified_if_needed(object, request, combined);
712 add_sampling_if_needed(object, combined);
713 add_analysis_outputs_if_needed(object, request, artifacts);
714 }
715
716 fn query_histogram_artifact(
717 &self,
718 request: &NetdataRequest,
719 combined: &CombinedResult,
720 context: &DisplayContext,
721 ) -> Option<Value> {
722 if let Some(histogram) = combined.histogram.as_ref() {
723 return Some(self.build_histogram(
724 context,
725 histogram,
726 combined.facets.get(&histogram.field),
727 ));
728 }
729 let histogram_field = request.histogram.as_ref()?;
730 if request.data_only && !request.delta {
731 return None;
732 }
733 let query = request.to_explorer_query(
734 combined.matched_files,
735 None,
736 NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC,
737 );
738 let histogram = empty_histogram_for_query(histogram_field.as_bytes(), &query);
739 Some(self.build_histogram(
740 context,
741 &histogram,
742 combined.facets.get(histogram.field.as_slice()),
743 ))
744 }
745
746 fn add_full_query_response_metadata(
747 &self,
748 object: &mut Map<String, Value>,
749 request: &NetdataRequest,
750 combined: &CombinedResult,
751 artifacts: &QueryResponseArtifacts,
752 ) {
753 object.insert("message".to_string(), artifacts.message.clone());
754 object.insert("update_every".to_string(), Value::from(1));
755 object.insert("help".to_string(), Value::Null);
756 object.insert(
757 "accepted_params".to_string(),
758 self.accepted_params_from_fields(&artifacts.reportable_facet_field_names),
759 );
760 object.insert("default_sort_column".to_string(), Value::from("timestamp"));
761 object.insert("default_charts".to_string(), Value::Array(Vec::new()));
762 object.insert(
763 "available_histograms".to_string(),
764 self.available_histograms(request, combined),
765 );
766 }
767
768 fn build_columns(
769 &self,
770 request: &NetdataRequest,
771 reportable_facet_fields: &[Vec<u8>],
772 rows: &[LocatedRow],
773 facets: &BTreeMap<Vec<u8>, BTreeMap<Vec<u8>, u64>>,
774 column_fields: &BTreeSet<String>,
775 ) -> Columns {
776 let mut order = vec!["timestamp".to_string(), "rowOptions".to_string()];
777 push_unique_many(&mut order, &self.config.default_view_keys);
778 push_unique_many(&mut order, &string_fields(reportable_facet_fields));
779 if let Some(histogram) = &request.histogram {
780 push_unique(&mut order, histogram);
781 }
782 for field in column_fields {
783 push_unique(&mut order, field);
784 }
785
786 for (field, values) in facets {
787 if !facet_group_is_reportable(values) {
788 continue;
789 }
790 push_unique(&mut order, &String::from_utf8_lossy(field));
791 }
792 for row in rows {
793 let fields = row_fields(row);
794 for field in fields.keys() {
795 push_unique(&mut order, field);
796 }
797 }
798
799 let mut map = Map::new();
800 for (index, key) in order.iter().enumerate() {
801 map.insert(key.clone(), column_metadata(key, index));
802 }
803 Columns { order, map }
804 }
805
806 fn build_data_rows(
807 &self,
808 context: &DisplayContext,
809 column_order: &[String],
810 rows: &[LocatedRow],
811 direction: Direction,
812 ) -> Vec<Value> {
813 let row_iter: Box<dyn Iterator<Item = &LocatedRow> + '_> = match direction {
814 Direction::Forward => Box::new(rows.iter().rev()),
815 Direction::Backward => Box::new(rows.iter()),
816 };
817 row_iter
818 .map(|located| {
819 let fields = row_fields(located);
820 let mut row = Vec::with_capacity(column_order.len());
821 for column in column_order {
822 let value = match column.as_str() {
823 "timestamp" => Value::from(located.row.realtime_usec),
824 "rowOptions" => self.profile.row_options(&fields),
825 field => first_value(&fields, field)
826 .map(|value| {
827 self.profile.field_display_value(
828 context,
829 DisplayScope::Data,
830 field,
831 value,
832 )
833 })
834 .unwrap_or(Value::Null),
835 };
836 row.push(value);
837 }
838 Value::Array(row)
839 })
840 .collect()
841 }
842
843 fn build_facets(
844 &self,
845 context: &DisplayContext,
846 requested: &[Vec<u8>],
847 facets: &BTreeMap<Vec<u8>, BTreeMap<Vec<u8>, u64>>,
848 ) -> Value {
849 let mut out = Vec::new();
850 for (order, field) in requested.iter().enumerate() {
851 let values = facets.get(field);
852 let field_name = String::from_utf8_lossy(field).into_owned();
853 let mut options: Vec<_> = values
854 .into_iter()
855 .flat_map(|values| values.iter())
856 .filter(|(value, count)| {
857 (!value.is_empty() && value.as_slice() != b"-")
858 || (**count == 0 && value.is_empty())
859 })
860 .map(|(value, count)| {
861 if *count == 0 && value.is_empty() {
862 return json!({
863 "id": NETDATA_EMPTY_STRING_FACET_HASH_ID,
864 "name": NETDATA_UNAVAILABLE_FIELD_LABEL,
865 "count": count,
866 });
867 }
868 json!({
869 "id": String::from_utf8_lossy(value).into_owned(),
870 "name": self.profile.facet_option_name(context, &field_name, value),
871 "count": count,
872 })
873 })
874 .collect();
875 sort_facet_options(&field_name, &mut options);
876 for (idx, option) in options.iter_mut().enumerate() {
877 if let Some(object) = option.as_object_mut() {
878 object.insert("order".to_string(), Value::from((idx + 1) as u64));
879 }
880 }
881 out.push(json!({
882 "id": field_name,
883 "name": String::from_utf8_lossy(field).into_owned(),
884 "order": order + 1,
885 "options": options,
886 }));
887 }
888 Value::Array(out)
889 }
890
891 fn build_histogram(
892 &self,
893 context: &DisplayContext,
894 histogram: &ExplorerHistogram,
895 known_values: Option<&BTreeMap<Vec<u8>, u64>>,
896 ) -> Value {
897 let field = String::from_utf8_lossy(&histogram.field).into_owned();
898 let mut dimension_ids = BTreeSet::new();
899 let mut buckets = Vec::with_capacity(histogram.buckets.len());
900 for bucket in &histogram.buckets {
901 let mut values = BTreeMap::new();
902 for (value, count) in &bucket.values {
903 add_netdata_facet_count(&mut values, value, *count);
904 }
905 for value in values.keys() {
906 dimension_ids.insert(value.clone());
907 }
908 buckets.push((bucket.start_realtime_usec, values));
909 }
910 let actual_dimension_ids = dimension_ids.clone();
911 if let Some(known_values) = known_values {
912 for value in known_values.keys() {
913 if value.is_empty() || value.as_slice() == b"-" {
914 continue;
915 }
916 dimension_ids.insert(value.clone());
917 }
918 }
919 let dimension_ids: Vec<Vec<u8>> = dimension_ids.into_iter().collect();
920 let metadata = self.histogram_chart_metadata(
921 context,
922 &field,
923 &buckets,
924 &actual_dimension_ids,
925 &dimension_ids,
926 );
927 let data: Vec<Value> = buckets
928 .iter()
929 .map(|(start_realtime_usec, values)| {
930 let mut point = Vec::with_capacity(dimension_ids.len() + 1);
931 point.push(Value::from(start_realtime_usec / 1000));
932 for value in &dimension_ids {
933 let count = values
934 .get(value)
935 .copied()
936 .map(Value::from)
937 .unwrap_or_else(|| {
938 if actual_dimension_ids.contains(value) {
939 Value::from(0)
940 } else {
941 Value::Null
942 }
943 });
944 point.push(Value::Array(vec![count, Value::from(0), Value::from(0)]));
945 }
946 Value::Array(point)
947 })
948 .collect();
949
950 json!({
951 "id": field,
952 "name": field,
953 "chart": {
954 "summary": metadata.summary,
955 "totals": metadata.totals,
956 "result": {
957 "labels": metadata.result_labels,
958 "point": { "value": 0, "arp": 1, "pa": 2 },
959 "data": data,
960 },
961 "db": {
962 "tiers": 1,
963 "update_every": histogram_update_every_seconds(histogram),
964 "units": "events",
965 "dimensions": {
966 "ids": metadata.ids.clone(),
967 "names": metadata.names.clone(),
968 "units": metadata.units.clone(),
969 "sts": metadata.stats.clone(),
970 },
971 "per_tier": [{
972 "tier": 0,
973 "queries": 1,
974 "points": metadata.points,
975 "update_every": histogram_update_every_seconds(histogram),
976 }],
977 },
978 "view": {
979 "title": format!("Events Distribution by {}", String::from_utf8_lossy(&histogram.field)),
980 "update_every": histogram_update_every_seconds(histogram),
981 "after": histogram_after_seconds(histogram),
982 "before": histogram_before_seconds(histogram),
983 "units": "events",
984 "chart_type": "stackedBar",
985 "dimensions": {
986 "grouped_by": ["dimension"],
987 "ids": metadata.ids,
988 "names": metadata.names,
989 "colors": metadata.colors,
990 "units": metadata.units,
991 "sts": metadata.stats,
992 },
993 "min": metadata.min,
994 "max": metadata.max,
995 },
996 "agents": [{
997 "mg": "default",
998 "nm": "facets.histogram",
999 "now": now_unix_seconds(),
1000 "ai": 0,
1001 }],
1002 }
1003 })
1004 }
1005
1006 fn histogram_chart_metadata(
1007 &self,
1008 context: &DisplayContext,
1009 field: &str,
1010 buckets: &[(u64, BTreeMap<Vec<u8>, u64>)],
1011 actual_dimensions: &BTreeSet<Vec<u8>>,
1012 dimensions: &[Vec<u8>],
1013 ) -> HistogramChartMetadata {
1014 let mut ids = Vec::with_capacity(dimensions.len());
1015 let mut names = Vec::with_capacity(dimensions.len());
1016 let mut colors = Vec::with_capacity(dimensions.len());
1017 let mut units = Vec::with_capacity(dimensions.len());
1018 let mut min_values = Vec::with_capacity(dimensions.len());
1019 let mut max_values = Vec::with_capacity(dimensions.len());
1020 let mut avg_values = Vec::with_capacity(dimensions.len());
1021 let mut arp_values = Vec::with_capacity(dimensions.len());
1022 let mut con_values = Vec::with_capacity(dimensions.len());
1023 let mut summary_dimensions = Vec::with_capacity(dimensions.len());
1024 let mut result_labels = vec![Value::String("time".to_string())];
1025
1026 let total: u64 = dimensions
1027 .iter()
1028 .map(|dimension| {
1029 buckets
1030 .iter()
1031 .map(|(_, values)| values.get(dimension).copied().unwrap_or(0))
1032 .sum::<u64>()
1033 })
1034 .sum();
1035
1036 let mut min = 0;
1037 let mut max = 0;
1038 let mut points = 0;
1039 for (priority, dimension) in dimensions.iter().enumerate() {
1040 let id = String::from_utf8_lossy(dimension).into_owned();
1041 let display = match self.profile.field_display_value(
1042 context,
1043 DisplayScope::Histogram,
1044 field,
1045 dimension,
1046 ) {
1047 Value::String(value) => value,
1048 other => other.to_string(),
1049 };
1050 let (dimension_min, dimension_max, dimension_sum, actual) =
1051 histogram_dimension_stats(buckets, actual_dimensions, dimension);
1052 let dimension_average = if actual && !buckets.is_empty() {
1053 dimension_sum as f64 / buckets.len() as f64
1054 } else {
1055 0.0
1056 };
1057 if actual {
1058 if points == 0 || dimension_min < min {
1059 min = dimension_min;
1060 }
1061 if dimension_max > max {
1062 max = dimension_max;
1063 }
1064 points += buckets.len() as u64;
1065 }
1066 let contribution = if total > 0 {
1067 dimension_sum as f64 * 100.0 / total as f64
1068 } else {
1069 0.0
1070 };
1071
1072 ids.push(Value::String(id.clone()));
1073 names.push(Value::String(display.clone()));
1074 colors.push(Value::Null);
1075 units.push(Value::String("events".to_string()));
1076 result_labels.push(Value::String(display.clone()));
1077 min_values.push(Value::from(dimension_min));
1078 max_values.push(Value::from(dimension_max));
1079 avg_values.push(Value::from(dimension_average));
1080 arp_values.push(Value::from(0));
1081 con_values.push(Value::from(contribution));
1082 summary_dimensions.push(json!({
1083 "id": id,
1084 "nm": display,
1085 "ds": { "sl": bool_to_u64(actual), "qr": bool_to_u64(actual) },
1086 "sts": {
1087 "min": dimension_min,
1088 "max": dimension_max,
1089 "avg": dimension_average,
1090 "con": contribution,
1091 },
1092 "pri": priority as u64,
1093 }));
1094 }
1095
1096 let stats = json!({
1097 "min": min_values,
1098 "max": max_values,
1099 "avg": avg_values,
1100 "arp": arp_values,
1101 "con": con_values,
1102 });
1103 let summary_stats = json!({
1104 "min": min,
1105 "max": max,
1106 "avg": histogram_average(total, points),
1107 "con": 100.0,
1108 });
1109 let dimensions_len = dimensions.len() as u64;
1110
1111 HistogramChartMetadata {
1112 ids,
1113 names,
1114 colors,
1115 units,
1116 stats,
1117 summary: json!({
1118 "nodes": [histogram_summary_node(dimensions_len, points, &summary_stats)],
1119 "contexts": [histogram_summary_context(dimensions_len, points, &summary_stats)],
1120 "instances": [histogram_summary_instance(dimensions_len, points, &summary_stats)],
1121 "dimensions": summary_dimensions,
1122 "labels": [],
1123 "alerts": [],
1124 }),
1125 totals: histogram_totals(dimensions_len),
1126 result_labels,
1127 min,
1128 max,
1129 points,
1130 }
1131 }
1132
1133 fn accepted_params_from_fields(&self, fields: &[String]) -> Value {
1134 NETDATA_ACCEPTED_PARAMS
1135 .iter()
1136 .copied()
1137 .chain(fields.iter().map(String::as_str))
1138 .map(|field| Value::String(field.to_string()))
1139 .collect()
1140 }
1141
1142 fn required_source_params(
1143 &self,
1144 paths: &[PathBuf],
1145 options: &NetdataFunctionRunOptions<'_>,
1146 ) -> Value {
1147 let mut all = JournalSourceSummary::default();
1148 let mut local = JournalSourceSummary::default();
1149 let mut local_namespaces = JournalSourceSummary::default();
1150 let mut local_system = JournalSourceSummary::default();
1151 let mut local_user = JournalSourceSummary::default();
1152 let mut remote = JournalSourceSummary::default();
1153 let mut other = JournalSourceSummary::default();
1154 let mut exact = BTreeMap::<String, JournalSourceSummary>::new();
1155
1156 for path in paths {
1157 let metadata = file_metadata(options, path);
1158 let source_type = metadata
1159 .as_ref()
1160 .and_then(|metadata| metadata.source_type)
1161 .unwrap_or_else(|| journal_file_source_type(path));
1162 all.add_path(path, self.config.reader_options, metadata.as_ref());
1163 if source_type & SOURCE_TYPE_LOCAL_ALL != 0 {
1164 local.add_path(path, self.config.reader_options, metadata.as_ref());
1165 }
1166 if source_type & SOURCE_TYPE_LOCAL_NAMESPACE != 0 {
1167 local_namespaces.add_path(path, self.config.reader_options, metadata.as_ref());
1168 }
1169 if source_type & SOURCE_TYPE_LOCAL_SYSTEM != 0 {
1170 local_system.add_path(path, self.config.reader_options, metadata.as_ref());
1171 }
1172 if source_type & SOURCE_TYPE_LOCAL_USER != 0 {
1173 local_user.add_path(path, self.config.reader_options, metadata.as_ref());
1174 }
1175 if source_type & SOURCE_TYPE_REMOTE_ALL != 0 {
1176 remote.add_path(path, self.config.reader_options, metadata.as_ref());
1177 }
1178 if source_type & SOURCE_TYPE_LOCAL_OTHER != 0 {
1179 other.add_path(path, self.config.reader_options, metadata.as_ref());
1180 }
1181 let source_name = metadata
1182 .as_ref()
1183 .and_then(|metadata| metadata.source_name.as_deref().map(str::to_owned))
1184 .or_else(|| journal_file_exact_source_name(path));
1185 if let Some(source_name) = source_name {
1186 exact.entry(source_name).or_default().add_path(
1187 path,
1188 self.config.reader_options,
1189 metadata.as_ref(),
1190 );
1191 }
1192 }
1193
1194 let mut source_options = Vec::new();
1195 push_source_option(&mut source_options, "all", &all);
1196 push_source_option(&mut source_options, "all-local-logs", &local);
1197 push_source_option(
1198 &mut source_options,
1199 "all-local-namespaces",
1200 &local_namespaces,
1201 );
1202 push_source_option(&mut source_options, "all-local-system-logs", &local_system);
1203 push_source_option(&mut source_options, "all-local-user-logs", &local_user);
1204 push_source_option(&mut source_options, "all-remote-systems", &remote);
1205 push_source_option(&mut source_options, "all-uncategorized", &other);
1206 for (name, summary) in exact {
1207 push_source_option(&mut source_options, &name, &summary);
1208 }
1209
1210 json!([{
1211 "id": "__logs_sources",
1212 "name": self.config.source_selector_name.as_str(),
1213 "help": self.config.source_selector_help.as_str(),
1214 "type": "multiselect",
1215 "options": source_options,
1216 }])
1217 }
1218
1219 fn available_histograms(&self, request: &NetdataRequest, combined: &CombinedResult) -> Value {
1220 let mut fields = combined.reportable_facet_fields(&request.facets);
1221 if request.data_only {
1222 if let Some(histogram) = &request.histogram {
1223 push_unique(&mut fields, histogram);
1224 }
1225 }
1226 let mut sorted = fields.clone();
1227 sorted.sort_by(|left, right| netdata_reorder_key(left).cmp(&netdata_reorder_key(right)));
1228 let order_by_field: BTreeMap<String, usize> = sorted
1229 .into_iter()
1230 .enumerate()
1231 .map(|(index, field)| (field, index + 1))
1232 .collect();
1233
1234 fields
1235 .into_iter()
1236 .map(|field| {
1237 let order = order_by_field.get(&field).copied().unwrap_or(0);
1238 json!({
1239 "id": field,
1240 "name": field,
1241 "order": order,
1242 })
1243 })
1244 .collect()
1245 }
1246}
1247
1248struct HistogramChartMetadata {
1249 ids: Vec<Value>,
1250 names: Vec<Value>,
1251 colors: Vec<Value>,
1252 units: Vec<Value>,
1253 stats: Value,
1254 summary: Value,
1255 totals: Value,
1256 result_labels: Vec<Value>,
1257 min: u64,
1258 max: u64,
1259 points: u64,
1260}
1261
1262fn histogram_dimension_stats(
1263 buckets: &[(u64, BTreeMap<Vec<u8>, u64>)],
1264 actual_dimensions: &BTreeSet<Vec<u8>>,
1265 dimension: &[u8],
1266) -> (u64, u64, u64, bool) {
1267 if !actual_dimensions.contains(dimension) {
1268 return (0, 0, 0, false);
1269 }
1270 let mut min = 0;
1271 let mut max = 0;
1272 let mut sum = 0;
1273 for (idx, (_, values)) in buckets.iter().enumerate() {
1274 let count = values.get(dimension).copied().unwrap_or(0);
1275 if idx == 0 || count < min {
1276 min = count;
1277 }
1278 if count > max {
1279 max = count;
1280 }
1281 sum += count;
1282 }
1283 (min, max, sum, true)
1284}
1285
1286fn histogram_average(total: u64, points: u64) -> f64 {
1287 if points == 0 {
1288 0.0
1289 } else {
1290 total as f64 / points as f64
1291 }
1292}
1293
1294fn histogram_summary_node(dimensions: u64, points: u64, stats: &Value) -> Value {
1295 let mut node = json!({
1296 "mg": "default",
1297 "nm": "facets.histogram",
1298 "ni": 0,
1299 "st": { "ai": 0, "code": 200, "msg": "" },
1300 });
1301 add_histogram_summary_shape(&mut node, dimensions, points, stats, true);
1302 node
1303}
1304
1305fn histogram_summary_context(dimensions: u64, points: u64, stats: &Value) -> Value {
1306 let mut context = json!({ "id": "facets.histogram" });
1307 add_histogram_summary_shape(&mut context, dimensions, points, stats, true);
1308 context
1309}
1310
1311fn histogram_summary_instance(dimensions: u64, points: u64, stats: &Value) -> Value {
1312 let mut instance = json!({ "id": "facets.histogram", "ni": 0 });
1313 add_histogram_summary_shape(&mut instance, dimensions, points, stats, false);
1314 instance
1315}
1316
1317fn add_histogram_summary_shape(
1318 object: &mut Value,
1319 dimensions: u64,
1320 points: u64,
1321 stats: &Value,
1322 include_instances: bool,
1323) {
1324 let Some(map) = object.as_object_mut() else {
1325 return;
1326 };
1327 if dimensions > 0 {
1328 if include_instances {
1329 map.insert("is".to_string(), json!({ "sl": 1, "qr": 1 }));
1330 }
1331 map.insert(
1332 "ds".to_string(),
1333 json!({ "sl": dimensions, "qr": dimensions }),
1334 );
1335 }
1336 if points > 0 {
1337 map.insert("sts".to_string(), stats.clone());
1338 }
1339}
1340
1341fn histogram_totals(dimensions: u64) -> Value {
1342 let mut totals = json!({ "nodes": { "sl": 1, "qr": 1 } });
1343 if dimensions > 0 {
1344 if let Some(map) = totals.as_object_mut() {
1345 map.insert("contexts".to_string(), json!({ "sl": 1, "qr": 1 }));
1346 map.insert("instances".to_string(), json!({ "sl": 1, "qr": 1 }));
1347 map.insert(
1348 "dimensions".to_string(),
1349 json!({ "sl": dimensions, "qr": dimensions }),
1350 );
1351 }
1352 }
1353 totals
1354}
1355
1356fn bool_to_u64(value: bool) -> u64 {
1357 if value { 1 } else { 0 }
1358}
1359
1360fn histogram_after_seconds(histogram: &ExplorerHistogram) -> u64 {
1361 histogram
1362 .buckets
1363 .first()
1364 .map(|bucket| bucket.start_realtime_usec / 1_000_000)
1365 .unwrap_or(0)
1366}
1367
1368fn histogram_before_seconds(histogram: &ExplorerHistogram) -> u64 {
1369 histogram
1370 .buckets
1371 .last()
1372 .map(|bucket| bucket.end_realtime_usec / 1_000_000)
1373 .unwrap_or(0)
1374}
1375
1376fn now_unix_seconds() -> u64 {
1377 SystemTime::now()
1378 .duration_since(UNIX_EPOCH)
1379 .unwrap_or_default()
1380 .as_secs()
1381}
1382
1383#[derive(Debug, Clone)]
1384struct NetdataRequest {
1385 info: bool,
1386 echo: Value,
1387 after_realtime_usec: Option<u64>,
1388 before_realtime_usec: Option<u64>,
1389 if_modified_since_usec: u64,
1390 anchor: ExplorerAnchor,
1391 direction: Direction,
1392 limit: usize,
1393 data_only: bool,
1394 delta: bool,
1395 tail: bool,
1396 sampling: u64,
1397 source_type: u64,
1398 exact_sources: Vec<String>,
1399 filters: Vec<ExplorerFilter>,
1400 facets: Vec<Vec<u8>>,
1401 histogram: Option<String>,
1402 fts_terms: Vec<ExplorerFtsPattern>,
1403 fts_patterns: Vec<Vec<u8>>,
1404 fts_negative_patterns: Vec<Vec<u8>>,
1405}
1406
1407impl NetdataRequest {
1408 fn parse(value: &Value, config: &NetdataFunctionConfig) -> Result<Self> {
1409 let object = value.as_object().ok_or_else(|| {
1410 SdkError::InvalidPath("Netdata function request must be a JSON object".to_string())
1411 })?;
1412 let now_seconds = unix_now_seconds();
1413 let info = get_bool(object, "info").unwrap_or(false);
1414 let after = get_i64(object, "after");
1415 let before = get_i64(object, "before");
1416 let (after_realtime_usec, before_realtime_usec) =
1417 normalize_time_window(now_seconds, after, before);
1418 let direction = request_direction(object);
1419 let if_modified_since_usec = get_u64(object, "if_modified_since").unwrap_or_default();
1420 let data_only = get_bool(object, "data_only").unwrap_or(false);
1421 let delta = request_delta(data_only, object);
1422 let tail = request_tail(data_only, if_modified_since_usec, object);
1423 let sampling = get_u64(object, "sampling").unwrap_or(DEFAULT_ITEMS_SAMPLING);
1424 let (anchor, direction) = request_anchor_and_direction(
1425 object,
1426 tail,
1427 direction,
1428 after_realtime_usec,
1429 before_realtime_usec,
1430 );
1431 let requested_limit = request_limit(object);
1432 let limit = requested_limit.max(2);
1433 let requested_facets = parse_string_array(object.get("facets"));
1434 let facets = request_facets(&requested_facets, config);
1435 let requested_histogram = request_histogram(object);
1436 let histogram = request_histogram_or_default(&requested_histogram, config);
1437 let requested_query = request_query(object);
1438 let (fts_terms, fts_patterns, fts_negative_patterns) = requested_query
1439 .as_deref()
1440 .map(parse_fts_query_patterns)
1441 .unwrap_or_default();
1442 let source_selection = parse_source_selection(object.get("selections"));
1443 let filters = parse_filters(object.get("selections"));
1444
1445 let echo_input = RequestEchoInput {
1446 info,
1447 after_realtime_usec,
1448 before_realtime_usec,
1449 if_modified_since_usec,
1450 anchor,
1451 direction,
1452 limit: requested_limit,
1453 data_only,
1454 delta,
1455 tail,
1456 sampling,
1457 source_type: source_selection.source_type,
1458 requested_facets: requested_facets.as_deref(),
1459 selections: object.get("selections"),
1460 histogram: requested_histogram.as_deref(),
1461 query: requested_query.as_deref(),
1462 };
1463 let echo = normalized_request_echo(&echo_input);
1464
1465 Ok(Self {
1466 info,
1467 echo,
1468 after_realtime_usec,
1469 before_realtime_usec,
1470 if_modified_since_usec,
1471 anchor,
1472 direction,
1473 limit,
1474 data_only,
1475 delta,
1476 tail,
1477 sampling,
1478 source_type: source_selection.source_type,
1479 exact_sources: source_selection.exact_sources,
1480 filters,
1481 facets,
1482 histogram,
1483 fts_terms,
1484 fts_patterns,
1485 fts_negative_patterns,
1486 })
1487 }
1488
1489 fn matches_source(&self, path: &Path, metadata: Option<&NetdataJournalFileMetadata>) -> bool {
1490 if self.source_type == SOURCE_TYPE_ALL && self.exact_sources.is_empty() {
1491 return true;
1492 }
1493 if self.source_type & SOURCE_TYPE_ALL != 0 {
1494 return true;
1495 }
1496 let file_source_type = metadata
1497 .and_then(|metadata| metadata.source_type)
1498 .unwrap_or_else(|| journal_file_source_type(path));
1499 if file_source_type & self.source_type != 0 {
1500 return true;
1501 }
1502 if self.exact_sources.is_empty() {
1503 return false;
1504 }
1505 let source_name = metadata
1506 .and_then(|metadata| metadata.source_name.as_deref().map(str::to_owned))
1507 .or_else(|| journal_file_exact_source_name(path));
1508 self.exact_sources
1509 .iter()
1510 .any(|source| source_name.as_deref() == Some(source.as_str()))
1511 }
1512
1513 fn to_explorer_query(
1514 &self,
1515 matched_files: u64,
1516 file_header: Option<FileHeader>,
1517 realtime_slack_usec: u64,
1518 ) -> ExplorerQuery {
1519 let analysis_enabled = !self.data_only || self.delta;
1520 let tail_anchor = self.tail && matches!(self.anchor, ExplorerAnchor::Realtime(_));
1521 let backward_page_anchor = !tail_anchor
1522 && self.direction == Direction::Backward
1523 && matches!(self.anchor, ExplorerAnchor::Realtime(_));
1524 let after_realtime_usec = if tail_anchor {
1525 tail_after_realtime_bound(self.after_realtime_usec, self.anchor)
1526 } else {
1527 self.after_realtime_usec
1528 };
1529 let before_realtime_usec = if backward_page_anchor {
1530 before_realtime_bound_excluding_anchor(self.before_realtime_usec, self.anchor)
1531 } else {
1532 self.before_realtime_usec
1533 };
1534 let anchor = if tail_anchor || backward_page_anchor {
1535 ExplorerAnchor::Auto
1536 } else {
1537 self.anchor
1538 };
1539 let sampling = (analysis_enabled
1540 && self.sampling != 0
1541 && matched_files != 0
1542 && after_realtime_usec.is_some()
1543 && before_realtime_usec.is_some())
1544 .then(|| {
1545 let header = file_header.unwrap_or(FileHeader {
1546 signature: [0; 8],
1547 compatible_flags: 0,
1548 incompatible_flags: 0,
1549 state: 0,
1550 file_id: [0; 16],
1551 machine_id: [0; 16],
1552 header_size: 0,
1553 arena_size: 0,
1554 data_hash_table_size: 0,
1555 field_hash_table_size: 0,
1556 n_objects: 0,
1557 n_entries: 0,
1558 head_entry_realtime: 0,
1559 tail_entry_realtime: 0,
1560 tail_entry_monotonic: 0,
1561 head_entry_seqnum: 0,
1562 tail_entry_seqnum: 0,
1563 tail_entry_boot_id: [0; 16],
1564 seqnum_id: [0; 16],
1565 n_data: 0,
1566 n_fields: 0,
1567 n_tags: 0,
1568 n_entry_arrays: 0,
1569 data_hash_chain_depth: 0,
1570 field_hash_chain_depth: 0,
1571 });
1572 let messages_in_file = header
1573 .tail_entry_seqnum
1574 .checked_sub(header.head_entry_seqnum)
1575 .and_then(|span| span.checked_add(1))
1576 .filter(|_| header.head_entry_seqnum != 0 && header.tail_entry_seqnum != 0)
1577 .unwrap_or(header.n_entries);
1578 ExplorerSampling {
1579 budget: self.sampling,
1580 matched_files,
1581 file_head_realtime_usec: header.head_entry_realtime,
1582 file_tail_realtime_usec: header.tail_entry_realtime,
1583 file_head_seqnum: header.head_entry_seqnum,
1584 file_tail_seqnum: header.tail_entry_seqnum,
1585 file_entries: messages_in_file,
1586 }
1587 });
1588 ExplorerQuery {
1589 after_realtime_usec,
1590 before_realtime_usec,
1591 anchor,
1592 direction: self.direction,
1593 limit: self.limit,
1594 filters: self.filters.clone(),
1595 facets: analysis_enabled
1596 .then(|| self.facets.clone())
1597 .unwrap_or_default(),
1598 histogram: analysis_enabled
1599 .then(|| {
1600 self.histogram
1601 .as_ref()
1602 .map(|field| field.as_bytes().to_vec())
1603 })
1604 .flatten(),
1605 histogram_after_realtime_usec: self.after_realtime_usec,
1606 histogram_before_realtime_usec: self.before_realtime_usec,
1607 histogram_target_buckets: DEFAULT_HISTOGRAM_BUCKETS,
1608 fts_terms: self.fts_terms.clone(),
1609 fts_patterns: self.fts_patterns.clone(),
1610 fts_negative_patterns: self.fts_negative_patterns.clone(),
1611 field_mode: ExplorerFieldMode::FirstValue,
1612 exclude_facet_field_filters: self.distinct_filter_fields() > 1,
1613 use_source_realtime: true,
1614 realtime_slack_usec: normalize_journal_vs_realtime_delta_usec(realtime_slack_usec),
1615 stop_when_rows_full: self.data_only && !tail_anchor,
1616 stop_when_rows_full_check_every: DATA_ONLY_CHECK_EVERY_ROWS,
1617 sampling,
1618 debug_collect_column_fields_by_row_traversal: false,
1619 }
1620 }
1621
1622 fn file_query(
1623 &self,
1624 matched_files: usize,
1625 file_header: FileHeader,
1626 order: &JournalFileOrderInfo,
1627 ) -> ExplorerQuery {
1628 let mut query = self.to_explorer_query(
1629 matched_files as u64,
1630 Some(file_header),
1631 order.journal_vs_realtime_delta_usec,
1632 );
1633 if self.data_only && self.delta {
1634 query.stop_when_rows_full = false;
1635 }
1636 query
1637 }
1638
1639 fn unfiltered_vocabulary(&self) -> Self {
1640 let mut request = self.clone();
1641 request.filters.clear();
1642 request.histogram = None;
1643 request.limit = 0;
1644 request.fts_terms.clear();
1645 request.fts_patterns.clear();
1646 request.fts_negative_patterns.clear();
1647 request
1648 }
1649
1650 fn distinct_filter_fields(&self) -> usize {
1651 self.filters
1652 .iter()
1653 .map(|filter| filter.field.as_slice())
1654 .collect::<HashSet<_>>()
1655 .len()
1656 }
1657}
1658
1659#[derive(Debug, Clone)]
1660struct LocatedRow {
1661 file_path: PathBuf,
1662 row: ExplorerRow,
1663}
1664
1665#[derive(Debug, Default)]
1666struct JournalSourceSummary {
1667 files: u64,
1668 total_size: u64,
1669 first_realtime_usec: Option<u64>,
1670 last_realtime_usec: Option<u64>,
1671}
1672
1673impl JournalSourceSummary {
1674 #[cfg(test)]
1675 fn from_paths(
1676 paths: &[PathBuf],
1677 reader_options: ReaderOptions,
1678 options: &NetdataFunctionRunOptions<'_>,
1679 ) -> Self {
1680 let mut summary = Self::default();
1681 for path in paths {
1682 let metadata = file_metadata(options, path);
1683 summary.add_path(path, reader_options, metadata.as_ref());
1684 }
1685 summary
1686 }
1687
1688 fn add_path(
1689 &mut self,
1690 path: &Path,
1691 reader_options: ReaderOptions,
1692 metadata: Option<&NetdataJournalFileMetadata>,
1693 ) {
1694 if let Ok(metadata) = std::fs::metadata(path) {
1695 self.files = self.files.saturating_add(1);
1696 self.total_size = self.total_size.saturating_add(metadata.len());
1697 }
1698 if let Some(metadata) = metadata {
1699 let metadata_first = metadata.msg_first_realtime_usec;
1700 let metadata_last = metadata.msg_last_realtime_usec;
1701 if let Some(first) = metadata_first {
1702 self.first_realtime_usec = Some(
1703 self.first_realtime_usec
1704 .map_or(first, |current| current.min(first)),
1705 );
1706 }
1707 if let Some(last) = metadata_last {
1708 self.last_realtime_usec = Some(
1709 self.last_realtime_usec
1710 .map_or(last, |current| current.max(last)),
1711 );
1712 }
1713 if metadata_first.is_some() && metadata_last.is_some() {
1714 return;
1715 }
1716 }
1717 let Ok(reader) = FileReader::open_with_options(path, reader_options) else {
1718 return;
1719 };
1720 let header = reader.header();
1721 if header.head_entry_realtime != 0 {
1722 self.first_realtime_usec = Some(
1723 self.first_realtime_usec
1724 .map_or(header.head_entry_realtime, |current| {
1725 current.min(header.head_entry_realtime)
1726 }),
1727 );
1728 }
1729 if header.tail_entry_realtime != 0 {
1730 self.last_realtime_usec = Some(
1731 self.last_realtime_usec
1732 .map_or(header.tail_entry_realtime, |current| {
1733 current.max(header.tail_entry_realtime)
1734 }),
1735 );
1736 }
1737 }
1738
1739 fn info(&self) -> String {
1740 let coverage = match (self.first_realtime_usec, self.last_realtime_usec) {
1741 (Some(first), Some(last)) if last > first && (last - first) >= 1_000_000 => {
1742 human_duration_seconds((last - first) / 1_000_000)
1743 }
1744 _ => "off".to_string(),
1745 };
1746 let last_entry = self
1747 .last_realtime_usec
1748 .and_then(|usec| DateTime::<Utc>::from_timestamp((usec / 1_000_000) as i64, 0))
1749 .map(|datetime| datetime.format("%Y-%m-%dT%H:%M:%SZ").to_string())
1750 .unwrap_or_else(|| "unknown".to_string());
1751 format!(
1752 "{} files, total size {}, covering {}, last entry at {}",
1753 self.files,
1754 human_binary_size(self.total_size),
1755 coverage,
1756 last_entry
1757 )
1758 }
1759}
1760
1761fn push_source_option(target: &mut Vec<Value>, id: &str, summary: &JournalSourceSummary) {
1762 if summary.files == 0 {
1763 return;
1764 }
1765 target.push(json!({
1766 "id": id,
1767 "name": id,
1768 "info": summary.info(),
1769 "pill": human_binary_size(summary.total_size),
1770 }));
1771}
1772
1773fn expand_located_row_payloads(
1774 located: &mut LocatedRow,
1775 reader_options: ReaderOptions,
1776) -> Result<()> {
1777 let mut reader = FileReader::open_with_options(&located.file_path, reader_options)?;
1778 reader.seek_cursor(&located.row.cursor)?;
1779 if !reader.test_cursor(&located.row.cursor)? {
1780 return Err(SdkError::InvalidCursor(format!(
1781 "selected row cursor is no longer available: {}",
1782 located.row.cursor
1783 )));
1784 }
1785 reader.collect_entry_payloads(&mut located.row.payloads)
1786}
1787
1788#[derive(Debug, Default)]
1789struct CombinedResult {
1790 rows: Vec<LocatedRow>,
1791 facets: BTreeMap<Vec<u8>, BTreeMap<Vec<u8>, u64>>,
1792 histogram: Option<ExplorerHistogram>,
1793 column_fields: BTreeSet<String>,
1794 stats: ExplorerStats,
1795 page_counters: Option<NetdataPageCounters>,
1796 matched_files: u64,
1797 matched_paths: Vec<PathBuf>,
1798 skipped_files: u64,
1799 file_errors: Vec<String>,
1800 partial: bool,
1801 timed_out: bool,
1802 cancelled: bool,
1803 sampling_enabled: bool,
1804}
1805
1806#[derive(Clone, Copy, Debug, Default)]
1807struct NetdataPageCounters {
1808 matched: u64,
1809 before: u64,
1810 after: u64,
1811}
1812
1813#[derive(Clone, Copy)]
1814struct ProgressContext {
1815 current_file: usize,
1816 total_files: usize,
1817 started: Instant,
1818}
1819
1820#[derive(Debug)]
1821enum NetdataPageHeap {
1822 Backward(BinaryHeap<Reverse<u64>>),
1823 Forward(BinaryHeap<u64>),
1824}
1825
1826#[derive(Debug)]
1827struct NetdataPageWindow {
1828 direction: Direction,
1829 anchor_start_usec: Option<u64>,
1830 anchor_stop_usec: Option<u64>,
1831 limit: usize,
1832 heap: NetdataPageHeap,
1833 oldest_retained_usec: Option<u64>,
1834 newest_retained_usec: Option<u64>,
1835 matched: u64,
1836 skips_before: u64,
1837 skips_after: u64,
1838 shifts: u64,
1839}
1840
1841impl NetdataPageWindow {
1842 fn for_request(request: &NetdataRequest) -> Self {
1843 let (anchor_start_usec, anchor_stop_usec) = match (request.tail, request.anchor) {
1844 (true, ExplorerAnchor::Realtime(anchor)) => (None, Some(anchor)),
1845 (false, ExplorerAnchor::Realtime(anchor)) => (Some(anchor), None),
1846 _ => (None, None),
1847 };
1848 let heap = match request.direction {
1849 Direction::Backward => NetdataPageHeap::Backward(BinaryHeap::new()),
1850 Direction::Forward => NetdataPageHeap::Forward(BinaryHeap::new()),
1851 };
1852 Self {
1853 direction: request.direction,
1854 anchor_start_usec,
1855 anchor_stop_usec,
1856 limit: request.limit,
1857 heap,
1858 oldest_retained_usec: None,
1859 newest_retained_usec: None,
1860 matched: 0,
1861 skips_before: 0,
1862 skips_after: if request.tail
1863 && request.delta
1864 && matches!(request.anchor, ExplorerAnchor::Realtime(_))
1865 {
1866 1
1867 } else {
1868 0
1869 },
1870 shifts: 0,
1871 }
1872 }
1873
1874 fn candidate_to_keep(&self, realtime_usec: u64) -> bool {
1875 if self.limit == 0 || !self.entry_within_anchor_readonly(realtime_usec) {
1876 return false;
1877 }
1878 if self.retained_len() < self.limit {
1879 return true;
1880 }
1881 self.oldest_retained_usec
1882 .zip(self.newest_retained_usec)
1883 .is_some_and(|(oldest, newest)| realtime_usec >= oldest && realtime_usec <= newest)
1884 }
1885
1886 fn observe(&mut self, realtime_usec: u64) {
1887 if !self.entry_within_anchor(realtime_usec) || self.limit == 0 {
1888 return;
1889 }
1890 self.matched = self.matched.saturating_add(1);
1891 match (&mut self.heap, self.direction) {
1892 (NetdataPageHeap::Backward(heap), Direction::Backward) => {
1893 if heap.len() < self.limit {
1894 heap.push(Reverse(realtime_usec));
1895 self.add_retained_bound(realtime_usec);
1896 return;
1897 }
1898 let Some(Reverse(oldest)) = heap.peek().copied() else {
1899 heap.push(Reverse(realtime_usec));
1900 self.refresh_retained_bounds();
1901 return;
1902 };
1903 if realtime_usec < oldest {
1904 self.skips_after = self.skips_after.saturating_add(1);
1905 return;
1906 }
1907 heap.pop();
1908 heap.push(Reverse(realtime_usec));
1909 self.refresh_retained_bounds();
1910 if realtime_usec > oldest {
1911 self.shifts = self.shifts.saturating_add(1);
1912 }
1913 }
1914 (NetdataPageHeap::Forward(heap), Direction::Forward) => {
1915 if heap.len() < self.limit {
1916 heap.push(realtime_usec);
1917 self.add_retained_bound(realtime_usec);
1918 return;
1919 }
1920 let Some(newest) = heap.peek().copied() else {
1921 heap.push(realtime_usec);
1922 self.refresh_retained_bounds();
1923 return;
1924 };
1925 if realtime_usec > newest {
1926 self.skips_before = self.skips_before.saturating_add(1);
1927 return;
1928 }
1929 heap.pop();
1930 heap.push(realtime_usec);
1931 self.refresh_retained_bounds();
1932 if realtime_usec < newest {
1933 self.shifts = self.shifts.saturating_add(1);
1934 }
1935 }
1936 _ => {}
1937 }
1938 }
1939
1940 fn retained_len(&self) -> usize {
1941 match &self.heap {
1942 NetdataPageHeap::Backward(heap) => heap.len(),
1943 NetdataPageHeap::Forward(heap) => heap.len(),
1944 }
1945 }
1946
1947 fn add_retained_bound(&mut self, realtime_usec: u64) {
1948 self.oldest_retained_usec = Some(
1949 self.oldest_retained_usec
1950 .map_or(realtime_usec, |current| current.min(realtime_usec)),
1951 );
1952 self.newest_retained_usec = Some(
1953 self.newest_retained_usec
1954 .map_or(realtime_usec, |current| current.max(realtime_usec)),
1955 );
1956 }
1957
1958 fn refresh_retained_bounds(&mut self) {
1959 let (oldest, newest) = match &self.heap {
1960 NetdataPageHeap::Backward(heap) => heap
1961 .iter()
1962 .map(|Reverse(usec)| *usec)
1963 .fold((None, None), retained_bounds_fold),
1964 NetdataPageHeap::Forward(heap) => heap
1965 .iter()
1966 .copied()
1967 .fold((None, None), retained_bounds_fold),
1968 };
1969 self.oldest_retained_usec = oldest;
1970 self.newest_retained_usec = newest;
1971 }
1972
1973 fn entry_within_anchor(&mut self, realtime_usec: u64) -> bool {
1974 match self.direction {
1975 Direction::Backward => {
1976 if self
1977 .anchor_start_usec
1978 .is_some_and(|anchor| realtime_usec >= anchor)
1979 {
1980 self.skips_before = self.skips_before.saturating_add(1);
1981 return false;
1982 }
1983 if self
1984 .anchor_stop_usec
1985 .is_some_and(|anchor| realtime_usec <= anchor)
1986 {
1987 self.skips_after = self.skips_after.saturating_add(1);
1988 return false;
1989 }
1990 }
1991 Direction::Forward => {
1992 if self
1993 .anchor_start_usec
1994 .is_some_and(|anchor| realtime_usec <= anchor)
1995 {
1996 self.skips_after = self.skips_after.saturating_add(1);
1997 return false;
1998 }
1999 if self
2000 .anchor_stop_usec
2001 .is_some_and(|anchor| realtime_usec >= anchor)
2002 {
2003 self.skips_before = self.skips_before.saturating_add(1);
2004 return false;
2005 }
2006 }
2007 }
2008 true
2009 }
2010
2011 fn entry_within_anchor_readonly(&self, realtime_usec: u64) -> bool {
2012 match self.direction {
2013 Direction::Backward => {
2014 if self
2015 .anchor_start_usec
2016 .is_some_and(|anchor| realtime_usec >= anchor)
2017 {
2018 return false;
2019 }
2020 if self
2021 .anchor_stop_usec
2022 .is_some_and(|anchor| realtime_usec <= anchor)
2023 {
2024 return false;
2025 }
2026 }
2027 Direction::Forward => {
2028 if self
2029 .anchor_start_usec
2030 .is_some_and(|anchor| realtime_usec <= anchor)
2031 {
2032 return false;
2033 }
2034 if self
2035 .anchor_stop_usec
2036 .is_some_and(|anchor| realtime_usec >= anchor)
2037 {
2038 return false;
2039 }
2040 }
2041 }
2042 true
2043 }
2044
2045 fn counters(&self) -> NetdataPageCounters {
2046 NetdataPageCounters {
2047 matched: self.matched,
2048 before: self.skips_before,
2049 after: self.skips_after.saturating_add(self.shifts),
2050 }
2051 }
2052
2053 fn can_stop_delta_file(&self, realtime_usec: u64, slack_usec: u64) -> bool {
2054 if self.limit == 0 {
2055 return false;
2056 }
2057 match (&self.heap, self.direction) {
2058 (NetdataPageHeap::Backward(heap), Direction::Backward) => {
2059 if heap.len() < self.limit {
2060 return false;
2061 }
2062 heap.peek().is_some_and(|Reverse(oldest)| {
2063 realtime_usec < oldest.saturating_sub(slack_usec)
2064 })
2065 }
2066 (NetdataPageHeap::Forward(heap), Direction::Forward) => {
2067 if heap.len() < self.limit {
2068 return false;
2069 }
2070 heap.peek()
2071 .is_some_and(|newest| realtime_usec > newest.saturating_add(slack_usec))
2072 }
2073 _ => false,
2074 }
2075 }
2076}
2077
2078fn retained_bounds_fold(
2079 (oldest, newest): (Option<u64>, Option<u64>),
2080 realtime_usec: u64,
2081) -> (Option<u64>, Option<u64>) {
2082 (
2083 Some(oldest.map_or(realtime_usec, |current| current.min(realtime_usec))),
2084 Some(newest.map_or(realtime_usec, |current| current.max(realtime_usec))),
2085 )
2086}
2087
2088impl CombinedResult {
2089 fn merge(
2090 &mut self,
2091 path: &Path,
2092 result: ExplorerResult,
2093 direction: Direction,
2094 limit: usize,
2095 ) -> Result<()> {
2096 let ExplorerResult {
2097 rows,
2098 facets,
2099 histogram,
2100 stats,
2101 column_fields,
2102 ..
2103 } = result;
2104
2105 if let Some(histogram) = histogram {
2106 merge_histogram(&mut self.histogram, histogram)?;
2107 }
2108
2109 self.merge_stats(stats);
2110 for row in rows {
2111 self.rows.push(LocatedRow {
2112 file_path: path.to_path_buf(),
2113 row,
2114 });
2115 }
2116 for field in column_fields {
2117 if let Ok(field) = String::from_utf8(field) {
2118 self.column_fields.insert(field);
2119 }
2120 }
2121 for (field, values) in facets {
2122 let target = self.facets.entry(field).or_default();
2123 for (value, count) in values {
2124 add_netdata_facet_count(target, &value, count);
2125 }
2126 }
2127 self.sort_and_limit(direction, limit);
2128 Ok(())
2129 }
2130
2131 fn add_column_fields<I>(&mut self, fields: I)
2132 where
2133 I: IntoIterator<Item = String>,
2134 {
2135 self.column_fields.extend(fields);
2136 }
2137
2138 fn sort_and_limit(&mut self, direction: Direction, limit: usize) {
2139 sort_rows_by_raw_realtime(&mut self.rows, direction);
2140 retain_boundary_group(&mut self.rows, direction, limit);
2141 make_row_timestamps_unique_same_file(&mut self.rows, direction);
2142 self.stats.rows_returned = self.rows.len() as u64;
2143 }
2144
2145 fn expand_row_payloads(&mut self, reader_options: ReaderOptions) {
2146 if self.rows.is_empty() {
2147 self.stats.rows_returned = 0;
2148 return;
2149 }
2150
2151 let mut rows = Vec::with_capacity(self.rows.len());
2152 for mut located in self.rows.drain(..) {
2153 if !located.row.payloads.is_empty() {
2154 rows.push(located);
2155 continue;
2156 }
2157 match expand_located_row_payloads(&mut located, reader_options) {
2158 Ok(()) => {
2159 self.stats.returned_row_expansions =
2160 self.stats.returned_row_expansions.saturating_add(1);
2161 rows.push(located);
2162 }
2163 Err(err) => {
2164 self.partial = true;
2165 self.file_errors.push(format!(
2166 "{} cursor {}: {err}",
2167 located.file_path.display(),
2168 located.row.cursor
2169 ));
2170 }
2171 }
2172 }
2173 self.rows = rows;
2174 self.stats.rows_returned = self.rows.len() as u64;
2175 }
2176
2177 fn merge_stats(&mut self, stats: ExplorerStats) {
2178 self.stats.rows_examined = self.stats.rows_examined.saturating_add(stats.rows_examined);
2179 self.stats.rows_matched = self.stats.rows_matched.saturating_add(stats.rows_matched);
2180 self.stats.facet_rows_matched = self
2181 .stats
2182 .facet_rows_matched
2183 .saturating_add(stats.facet_rows_matched);
2184 self.stats.rows_returned = self.stats.rows_returned.saturating_add(stats.rows_returned);
2185 self.stats.rows_unsampled = self
2186 .stats
2187 .rows_unsampled
2188 .saturating_add(stats.rows_unsampled);
2189 self.stats.rows_estimated = self
2190 .stats
2191 .rows_estimated
2192 .saturating_add(stats.rows_estimated);
2193 self.stats.sampling_sampled = self
2194 .stats
2195 .sampling_sampled
2196 .saturating_add(stats.sampling_sampled);
2197 self.stats.sampling_unsampled = self
2198 .stats
2199 .sampling_unsampled
2200 .saturating_add(stats.sampling_unsampled);
2201 self.stats.sampling_estimated = self
2202 .stats
2203 .sampling_estimated
2204 .saturating_add(stats.sampling_estimated);
2205 if stats.last_realtime_usec > self.stats.last_realtime_usec {
2206 self.stats.last_realtime_usec = stats.last_realtime_usec;
2207 }
2208 if stats.max_source_realtime_delta_usec > self.stats.max_source_realtime_delta_usec {
2209 self.stats.max_source_realtime_delta_usec = stats.max_source_realtime_delta_usec;
2210 }
2211 self.stats.data_refs_seen = self
2212 .stats
2213 .data_refs_seen
2214 .saturating_add(stats.data_refs_seen);
2215 self.stats.data_refs_skipped = self
2216 .stats
2217 .data_refs_skipped
2218 .saturating_add(stats.data_refs_skipped);
2219 self.stats.data_payloads_loaded = self
2220 .stats
2221 .data_payloads_loaded
2222 .saturating_add(stats.data_payloads_loaded);
2223 self.stats.data_objects_classified = self
2224 .stats
2225 .data_objects_classified
2226 .saturating_add(stats.data_objects_classified);
2227 self.stats.data_cache_hits = self
2228 .stats
2229 .data_cache_hits
2230 .saturating_add(stats.data_cache_hits);
2231 self.stats.data_cache_misses = self
2232 .stats
2233 .data_cache_misses
2234 .saturating_add(stats.data_cache_misses);
2235 self.stats.payloads_decompressed = self
2236 .stats
2237 .payloads_decompressed
2238 .saturating_add(stats.payloads_decompressed);
2239 self.stats.fts_scans = self.stats.fts_scans.saturating_add(stats.fts_scans);
2240 self.stats.facet_updates = self.stats.facet_updates.saturating_add(stats.facet_updates);
2241 self.stats.histogram_updates = self
2242 .stats
2243 .histogram_updates
2244 .saturating_add(stats.histogram_updates);
2245 self.stats.returned_row_expansions = self
2246 .stats
2247 .returned_row_expansions
2248 .saturating_add(stats.returned_row_expansions);
2249 self.stats.early_stop_opportunities = self
2250 .stats
2251 .early_stop_opportunities
2252 .saturating_add(stats.early_stop_opportunities);
2253 self.stats.early_stops = self.stats.early_stops.saturating_add(stats.early_stops);
2254 }
2255
2256 fn add_zero_count_facet_values(
2257 &mut self,
2258 vocabulary: &BTreeMap<Vec<u8>, BTreeMap<Vec<u8>, u64>>,
2259 ) {
2260 for (field, values) in vocabulary {
2261 let target = self.facets.entry(field.clone()).or_default();
2262 for value in values.keys() {
2263 add_netdata_facet_count(target, value, 0);
2264 }
2265 }
2266 }
2267
2268 fn add_zero_count_facet_values_from_files(
2269 &mut self,
2270 fields: &[Vec<u8>],
2271 reader_options: ReaderOptions,
2272 ) {
2273 for path in &self.matched_paths {
2274 let Ok(mut reader) = FileReader::open_with_options(path, reader_options) else {
2275 continue;
2276 };
2277 for field in fields {
2278 let Ok(field_name) = std::str::from_utf8(field) else {
2279 continue;
2280 };
2281 let Ok(values) = reader.query_unique(field_name) else {
2282 continue;
2283 };
2284 if values.is_empty() {
2285 continue;
2286 }
2287 let target = self.facets.entry(field.clone()).or_default();
2288 for value in values {
2289 add_netdata_facet_count(target, &value, 0);
2290 }
2291 }
2292 }
2293 }
2294
2295 fn add_zero_count_selected_filter_values(&mut self, request: &NetdataRequest) {
2296 let mut report_fields: HashSet<Vec<u8>> = request.facets.iter().cloned().collect();
2297 if let Some(histogram) = &request.histogram {
2298 report_fields.insert(histogram.as_bytes().to_vec());
2299 }
2300 for filter in &request.filters {
2301 if !report_fields.contains(&filter.field) {
2302 continue;
2303 }
2304 let target = self.facets.entry(filter.field.clone()).or_default();
2305 for value in &filter.values {
2306 add_netdata_facet_count(target, value, 0);
2307 }
2308 }
2309 }
2310
2311 fn reportable_facet_fields(&self, requested: &[Vec<u8>]) -> Vec<String> {
2312 string_fields(&self.reportable_facet_fields_bytes(requested))
2313 }
2314
2315 fn reportable_facet_fields_bytes(&self, requested: &[Vec<u8>]) -> Vec<Vec<u8>> {
2316 let mut fields = Vec::new();
2317 for field in requested {
2318 if !fields.contains(field) {
2319 fields.push(field.clone());
2320 }
2321 }
2322 fields
2323 }
2324}
2325
2326fn netdata_function_error(status: u64, message: &str) -> Value {
2327 json!({
2328 "status": status,
2329 "errorMessage": message,
2330 })
2331}
2332
2333fn query_message(timed_out: bool, stats: &ExplorerStats) -> Value {
2334 if !timed_out && stats.rows_unsampled == 0 && stats.rows_estimated == 0 {
2335 return Value::String("OK".to_string());
2336 }
2337
2338 let total = stats
2339 .rows_examined
2340 .saturating_add(stats.rows_unsampled)
2341 .saturating_add(stats.rows_estimated)
2342 .max(1);
2343 let real_percent = stats.rows_examined as f64 * 100.0 / total as f64;
2344 let unsampled_percent = stats.rows_unsampled as f64 * 100.0 / total as f64;
2345 let estimated_percent = stats.rows_estimated as f64 * 100.0 / total as f64;
2346
2347 let mut title = String::new();
2348 let mut description = String::new();
2349 let mut status = "notice";
2350 if timed_out {
2351 title.push_str("Query timed-out, incomplete data. ");
2352 description.push_str(
2353 "QUERY TIMEOUT: The query timed out and may not include all the data of the selected window. ",
2354 );
2355 status = "warning";
2356 }
2357 if stats.rows_unsampled != 0 || stats.rows_estimated != 0 {
2358 title.push_str(&format!("{real_percent:.2}% real data"));
2359 description.push_str(&format!(
2360 "ACTUAL DATA: The filters counters reflect {real_percent:.2}% of the data. "
2361 ));
2362 }
2363 if stats.rows_unsampled != 0 {
2364 title.push_str(&format!(", {unsampled_percent:.2}% unsampled"));
2365 description.push_str(&format!(
2366 "UNSAMPLED DATA: {unsampled_percent:.2}% of the events exist and have been counted, but their values have not been evaluated, so they are not included in the filters counters. "
2367 ));
2368 }
2369 if stats.rows_estimated != 0 {
2370 title.push_str(&format!(", {estimated_percent:.2}% estimated"));
2371 description.push_str(&format!(
2372 "ESTIMATED DATA: The query selected a large amount of data, so to avoid delaying too much, the presented data are estimated by {estimated_percent:.2}%. "
2373 ));
2374 }
2375
2376 json!({
2377 "title": title,
2378 "status": status,
2379 "description": description,
2380 })
2381}
2382
2383fn merged_progress_stats(completed: &ExplorerStats, current: &ExplorerStats) -> ExplorerStats {
2384 let mut stats = completed.clone();
2385 stats.rows_examined = stats.rows_examined.saturating_add(current.rows_examined);
2386 stats.rows_matched = stats.rows_matched.saturating_add(current.rows_matched);
2387 stats.facet_rows_matched = stats
2388 .facet_rows_matched
2389 .saturating_add(current.facet_rows_matched);
2390 stats.rows_returned = stats.rows_returned.saturating_add(current.rows_returned);
2391 stats.rows_unsampled = stats.rows_unsampled.saturating_add(current.rows_unsampled);
2392 stats.rows_estimated = stats.rows_estimated.saturating_add(current.rows_estimated);
2393 stats.sampling_sampled = stats
2394 .sampling_sampled
2395 .saturating_add(current.sampling_sampled);
2396 stats.sampling_unsampled = stats
2397 .sampling_unsampled
2398 .saturating_add(current.sampling_unsampled);
2399 stats.sampling_estimated = stats
2400 .sampling_estimated
2401 .saturating_add(current.sampling_estimated);
2402 if current.last_realtime_usec > stats.last_realtime_usec {
2403 stats.last_realtime_usec = current.last_realtime_usec;
2404 }
2405 if current.max_source_realtime_delta_usec > stats.max_source_realtime_delta_usec {
2406 stats.max_source_realtime_delta_usec = current.max_source_realtime_delta_usec;
2407 }
2408 stats.data_refs_seen = stats.data_refs_seen.saturating_add(current.data_refs_seen);
2409 stats.data_refs_skipped = stats
2410 .data_refs_skipped
2411 .saturating_add(current.data_refs_skipped);
2412 stats.data_payloads_loaded = stats
2413 .data_payloads_loaded
2414 .saturating_add(current.data_payloads_loaded);
2415 stats.data_objects_classified = stats
2416 .data_objects_classified
2417 .saturating_add(current.data_objects_classified);
2418 stats.data_cache_hits = stats
2419 .data_cache_hits
2420 .saturating_add(current.data_cache_hits);
2421 stats.data_cache_misses = stats
2422 .data_cache_misses
2423 .saturating_add(current.data_cache_misses);
2424 stats.payloads_decompressed = stats
2425 .payloads_decompressed
2426 .saturating_add(current.payloads_decompressed);
2427 stats.fts_scans = stats.fts_scans.saturating_add(current.fts_scans);
2428 stats.facet_updates = stats.facet_updates.saturating_add(current.facet_updates);
2429 stats.histogram_updates = stats
2430 .histogram_updates
2431 .saturating_add(current.histogram_updates);
2432 stats.returned_row_expansions = stats
2433 .returned_row_expansions
2434 .saturating_add(current.returned_row_expansions);
2435 stats.early_stop_opportunities = stats
2436 .early_stop_opportunities
2437 .saturating_add(current.early_stop_opportunities);
2438 stats.early_stops = stats.early_stops.saturating_add(current.early_stops);
2439 stats
2440}
2441
2442struct Columns {
2443 order: Vec<String>,
2444 map: Map<String, Value>,
2445}
2446
2447struct QueryResponseArtifacts {
2448 reportable_facet_field_names: Vec<String>,
2449 columns: Columns,
2450 data: Vec<Value>,
2451 facets: Value,
2452 histogram: Option<Value>,
2453 message: Value,
2454 items: Value,
2455}
2456
2457fn base_query_response(
2458 request: &NetdataRequest,
2459 combined: &CombinedResult,
2460 artifacts: &QueryResponseArtifacts,
2461) -> Value {
2462 json!({
2463 "_request": &request.echo,
2464 "versions": { "netdata_function_api": 1, "sdk": env!("CARGO_PKG_VERSION") },
2465 "_journal_files": {
2466 "matched": combined.matched_files,
2467 "skipped": combined.skipped_files,
2468 "errors": &combined.file_errors,
2469 },
2470 "status": 200,
2471 "partial": combined.partial,
2472 "type": "table",
2473 "show_ids": true,
2474 "has_history": true,
2475 "pagination": {
2476 "enabled": true,
2477 "key": "anchor",
2478 "column": "timestamp",
2479 "units": "timestamp_usec",
2480 },
2481 "columns": &artifacts.columns.map,
2482 "data": &artifacts.data,
2483 "_stats": {
2484 "sdk_explorer": &combined.stats,
2485 },
2486 "expires": if request.data_only {
2487 unix_now_seconds().saturating_add(3600)
2488 } else {
2489 0
2490 }
2491 })
2492}
2493
2494fn response_items(request: &NetdataRequest, combined: &CombinedResult, returned: u64) -> Value {
2495 let unsampled = combined.stats.rows_unsampled;
2496 let estimated = combined.stats.rows_estimated;
2497 let fallback_rows_after_returned =
2498 response_fallback_rows_after_returned(&combined.stats, returned);
2499 let page_counters = combined
2500 .page_counters
2501 .unwrap_or_else(|| NetdataPageCounters {
2502 matched: combined.stats.rows_matched,
2503 before: 0,
2504 after: fallback_rows_after_returned,
2505 });
2506 json!({
2507 "evaluated": combined.stats.rows_examined.saturating_add(unsampled).saturating_add(estimated),
2508 "matched": page_counters.matched.saturating_add(unsampled).saturating_add(estimated),
2509 "unsampled": unsampled,
2510 "estimated": estimated,
2511 "returned": returned,
2512 "max_to_return": request.limit as u64,
2513 "before": page_counters.before,
2514 "after": page_counters.after,
2515 })
2516}
2517
2518fn response_fallback_rows_after_returned(stats: &ExplorerStats, returned: u64) -> u64 {
2519 let source_rows = if stats.rows_unsampled != 0 || stats.rows_estimated != 0 {
2520 stats.rows_examined
2521 } else {
2522 stats.rows_matched
2523 };
2524 source_rows.saturating_sub(returned)
2525}
2526
2527fn add_last_modified_if_needed(
2528 object: &mut Map<String, Value>,
2529 request: &NetdataRequest,
2530 combined: &CombinedResult,
2531) {
2532 if !request.data_only || request.tail {
2533 object.insert(
2534 "last_modified".to_string(),
2535 Value::from(combined.stats.last_realtime_usec),
2536 );
2537 }
2538}
2539
2540fn add_sampling_if_needed(object: &mut Map<String, Value>, combined: &CombinedResult) {
2541 if combined.sampling_enabled {
2542 object.insert(
2543 "_sampling".to_string(),
2544 json!({
2545 "enabled": true,
2546 "sampled": combined.stats.sampling_sampled,
2547 "unsampled": combined.stats.sampling_unsampled,
2548 "estimated": combined.stats.sampling_estimated,
2549 }),
2550 );
2551 }
2552}
2553
2554fn add_analysis_outputs_if_needed(
2555 object: &mut Map<String, Value>,
2556 request: &NetdataRequest,
2557 artifacts: QueryResponseArtifacts,
2558) {
2559 if !request.data_only || request.delta {
2560 let (facets_key, histogram_key, items_key) = response_analysis_keys(request.data_only);
2561 object.insert(facets_key.to_string(), artifacts.facets);
2562 object.insert(
2563 histogram_key.to_string(),
2564 artifacts.histogram.unwrap_or(Value::Null),
2565 );
2566 object.insert(items_key.to_string(), artifacts.items);
2567 }
2568}
2569
2570fn response_analysis_keys(data_only: bool) -> (&'static str, &'static str, &'static str) {
2571 if data_only {
2572 ("facets_delta", "histogram_delta", "items_delta")
2573 } else {
2574 ("facets", "histogram", "items")
2575 }
2576}
2577
2578fn merge_histogram(
2579 target: &mut Option<ExplorerHistogram>,
2580 source: ExplorerHistogram,
2581) -> Result<()> {
2582 let Some(target) = target else {
2583 *target = Some(source);
2584 return Ok(());
2585 };
2586 if target.field != source.field
2587 || target.buckets.len() != source.buckets.len()
2588 || target
2589 .buckets
2590 .iter()
2591 .zip(source.buckets.iter())
2592 .any(|(target_bucket, source_bucket)| {
2593 target_bucket.start_realtime_usec != source_bucket.start_realtime_usec
2594 || target_bucket.end_realtime_usec != source_bucket.end_realtime_usec
2595 })
2596 {
2597 return Err(SdkError::Unsupported(
2598 "inconsistent Netdata histogram bucket shape",
2599 ));
2600 }
2601 for (index, source_bucket) in source.buckets.into_iter().enumerate() {
2602 let Some(target_bucket) = target.buckets.get_mut(index) else {
2603 return Err(SdkError::Unsupported(
2604 "inconsistent Netdata histogram bucket shape",
2605 ));
2606 };
2607 for (value, count) in source_bucket.values {
2608 *target_bucket.values.entry(value).or_default() += count;
2609 }
2610 }
2611 Ok(())
2612}
2613
2614fn facet_group_is_reportable(values: &BTreeMap<Vec<u8>, u64>) -> bool {
2615 values
2616 .iter()
2617 .any(|(value, _count)| !value.is_empty() && value.as_slice() != b"-")
2618}
2619
2620fn netdata_facet_value(value: &[u8]) -> &[u8] {
2621 if value.len() > NETDATA_FACET_MAX_VALUE_LENGTH {
2622 &value[..NETDATA_FACET_MAX_VALUE_LENGTH]
2623 } else {
2624 value
2625 }
2626}
2627
2628fn add_netdata_facet_count(target: &mut BTreeMap<Vec<u8>, u64>, value: &[u8], count: u64) {
2629 *target
2630 .entry(netdata_facet_value(value).to_vec())
2631 .or_default() += count;
2632}
2633
2634fn not_modified_before_scan_response(
2635 request: &NetdataRequest,
2636 selected: &SelectedJournalFiles,
2637) -> Option<Value> {
2638 if request.if_modified_since_usec != 0 && !selected.files_are_newer {
2639 Some(netdata_function_error(
2640 304,
2641 "No new data since the previous call.",
2642 ))
2643 } else {
2644 None
2645 }
2646}
2647
2648fn should_collect_unfiltered_facet_vocabulary(
2649 request: &NetdataRequest,
2650 combined: &CombinedResult,
2651) -> bool {
2652 !request.data_only && !combined.partial && !request.filters.is_empty()
2653}
2654
2655fn progress_context(file_index: usize, total_files: usize, started: Instant) -> ProgressContext {
2656 ProgressContext {
2657 current_file: file_index + 1,
2658 total_files,
2659 started,
2660 }
2661}
2662
2663fn emit_netdata_progress(
2664 options: &mut NetdataFunctionRunOptions<'_>,
2665 progress: NetdataFunctionProgress,
2666) {
2667 if let Some(callback) = options.progress_callback.as_deref_mut() {
2668 callback(progress);
2669 }
2670}
2671
2672fn emit_progress_for_combined(
2673 options: &mut NetdataFunctionRunOptions<'_>,
2674 combined: &CombinedResult,
2675 context: ProgressContext,
2676) {
2677 emit_netdata_progress(
2678 options,
2679 NetdataFunctionProgress {
2680 current_file: context.current_file,
2681 total_files: context.total_files,
2682 matched_files: combined.matched_files,
2683 skipped_files: combined.skipped_files,
2684 stats: combined.stats.clone(),
2685 elapsed: context.started.elapsed(),
2686 },
2687 );
2688}
2689
2690fn emit_explorer_progress(
2691 options: &mut NetdataFunctionRunOptions<'_>,
2692 combined: &CombinedResult,
2693 progress: ExplorerProgress,
2694 context: ProgressContext,
2695) {
2696 let stats = merged_progress_stats(&combined.stats, &progress.stats);
2697 if let Some(callback) = options.progress_callback.as_deref_mut() {
2698 callback(NetdataFunctionProgress {
2699 current_file: context.current_file,
2700 total_files: context.total_files,
2701 matched_files: combined.matched_files,
2702 skipped_files: combined.skipped_files,
2703 stats,
2704 elapsed: context.started.elapsed(),
2705 });
2706 }
2707}
2708
2709fn request_cancelled(options: &NetdataFunctionRunOptions<'_>) -> bool {
2710 options
2711 .cancellation_callback
2712 .is_some_and(|is_cancelled| is_cancelled())
2713}
2714
2715fn should_stop_before_file(
2716 combined: &mut CombinedResult,
2717 deadline: Option<Instant>,
2718 options: &NetdataFunctionRunOptions<'_>,
2719) -> bool {
2720 if request_cancelled(options) {
2721 combined.partial = true;
2722 combined.cancelled = true;
2723 return true;
2724 }
2725 if deadline.is_some_and(|deadline| Instant::now() >= deadline) {
2726 combined.partial = true;
2727 combined.timed_out = true;
2728 return true;
2729 }
2730 false
2731}
2732
2733fn collect_column_fields_for_file(
2734 reader: &mut FileReader,
2735 request: &NetdataRequest,
2736 path: &Path,
2737 combined: &mut CombinedResult,
2738) {
2739 if request.data_only {
2740 return;
2741 }
2742 match reader.enumerate_fields_indexed() {
2743 Ok(fields) => combined.add_column_fields(fields),
2744 Err(err) => combined.file_errors.push(format!(
2745 "{}: FIELD index enumeration failed: {err}",
2746 path.display()
2747 )),
2748 }
2749}
2750
2751fn record_explore_result(
2752 result: Result<(ExplorerResult, Option<ExplorerStopReason>)>,
2753 path: &Path,
2754 combined: &mut CombinedResult,
2755) -> Option<(ExplorerResult, Option<ExplorerStopReason>)> {
2756 match result {
2757 Ok(result) => Some(result),
2758 Err(err) => {
2759 combined.skipped_files = combined.skipped_files.saturating_add(1);
2760 combined
2761 .file_errors
2762 .push(format!("{}: {err}", path.display()));
2763 None
2764 }
2765 }
2766}
2767
2768fn delta_scan_can_stop(
2769 request: &NetdataRequest,
2770 page_window: &RefCell<NetdataPageWindow>,
2771 realtime_usec: u64,
2772 rows_matched: u64,
2773 realtime_delta_usec: u64,
2774) -> bool {
2775 let mut page_window = page_window.borrow_mut();
2776 page_window.observe(realtime_usec);
2777 request.data_only
2778 && request.delta
2779 && rows_matched % DATA_ONLY_CHECK_EVERY_ROWS == 0
2780 && page_window.can_stop_delta_file(realtime_usec, realtime_delta_usec)
2781}
2782
2783#[allow(clippy::too_many_arguments)]
2784fn finish_explored_file(
2785 options: &mut NetdataFunctionRunOptions<'_>,
2786 request: &NetdataRequest,
2787 file: &SelectedJournalFile,
2788 query: &ExplorerQuery,
2789 result: ExplorerResult,
2790 stop_reason: Option<ExplorerStopReason>,
2791 combined: &mut CombinedResult,
2792 files: &[SelectedJournalFile],
2793 file_index: usize,
2794 progress: ProgressContext,
2795) -> Result<bool> {
2796 update_learned_realtime_delta(options, &file.path, &file.order, &result.stats);
2797 combined.merge(&file.path, result, query.direction, request.limit)?;
2798 emit_progress_for_combined(options, combined, progress);
2799 if request_cancelled(options) {
2800 combined.partial = true;
2801 combined.cancelled = true;
2802 return Ok(true);
2803 }
2804 if let Some(reason) = stop_reason {
2805 combined.partial = true;
2806 match reason {
2807 ExplorerStopReason::TimedOut => combined.timed_out = true,
2808 ExplorerStopReason::Cancelled => combined.cancelled = true,
2809 }
2810 return Ok(true);
2811 }
2812 Ok(request.data_only
2813 && !request.delta
2814 && !request.tail
2815 && combined.rows.len() >= request.limit
2816 && remaining_files_cannot_affect_data_page(combined, request, files, file_index + 1))
2817}
2818
2819fn file_metadata(
2820 options: &NetdataFunctionRunOptions<'_>,
2821 path: &Path,
2822) -> Option<NetdataJournalFileMetadata> {
2823 options
2824 .state
2825 .as_deref()
2826 .and_then(|state| state.file_metadata(path))
2827}
2828
2829fn update_learned_realtime_delta(
2830 options: &mut NetdataFunctionRunOptions<'_>,
2831 path: &Path,
2832 order: &JournalFileOrderInfo,
2833 stats: &ExplorerStats,
2834) {
2835 let learned_delta = stats.max_source_realtime_delta_usec;
2836 if learned_delta == 0 || learned_delta <= order.journal_vs_realtime_delta_usec {
2837 return;
2838 }
2839 let learned_delta = normalize_journal_vs_realtime_delta_usec(learned_delta);
2840 if learned_delta <= order.journal_vs_realtime_delta_usec {
2841 return;
2842 }
2843 if let Some(state) = options.state.as_deref_mut() {
2844 state.update_file_journal_vs_realtime_delta_usec(path, learned_delta);
2845 }
2846}
2847
2848fn normalize_journal_vs_realtime_delta_usec(delta_usec: u64) -> u64 {
2849 delta_usec
2850 .max(NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC)
2851 .min(NETDATA_JOURNAL_VS_REALTIME_DELTA_MAX_USEC)
2852}
2853
2854#[cfg(test)]
2855fn file_may_overlap_request(header: crate::FileHeader, request: &NetdataRequest) -> bool {
2856 if header.tail_entry_realtime == 0 {
2857 return true;
2858 }
2859
2860 let first = header
2861 .head_entry_realtime
2862 .saturating_sub(NETDATA_JOURNAL_VS_REALTIME_DELTA_MAX_USEC);
2863 let last = header
2864 .tail_entry_realtime
2865 .saturating_add(NETDATA_JOURNAL_VS_REALTIME_DELTA_MAX_USEC);
2866
2867 if request
2868 .after_realtime_usec
2869 .is_some_and(|after| last < after)
2870 {
2871 return false;
2872 }
2873 if request
2874 .before_realtime_usec
2875 .is_some_and(|before| first > before)
2876 {
2877 return false;
2878 }
2879
2880 true
2881}
2882
2883#[derive(Debug)]
2884struct SelectedJournalFile {
2885 path: PathBuf,
2886 order: JournalFileOrderInfo,
2887}
2888
2889#[derive(Debug, Default)]
2890struct SelectedJournalFiles {
2891 files: Vec<SelectedJournalFile>,
2892 files_are_newer: bool,
2893}
2894
2895fn select_journal_files_for_request(
2896 paths: Vec<PathBuf>,
2897 request: &NetdataRequest,
2898 reader_options: ReaderOptions,
2899 options: &NetdataFunctionRunOptions<'_>,
2900) -> SelectedJournalFiles {
2901 let mut selected = Vec::new();
2902 for path in paths {
2903 let metadata = file_metadata(options, &path);
2904 if !request.matches_source(&path, metadata.as_ref()) {
2905 continue;
2906 }
2907 let order = journal_file_order_info(&path, reader_options, metadata.as_ref());
2908 if !journal_file_order_may_overlap_request(&order, request) {
2909 continue;
2910 }
2911 selected.push(SelectedJournalFile { path, order });
2912 }
2913 selected.sort_by(|left, right| {
2914 compare_journal_file_order(&left.order, &right.order, request.direction)
2915 .then_with(|| left.path.cmp(&right.path))
2916 });
2917 let files_are_newer = selected
2918 .iter()
2919 .any(|file| file.order.msg_last_realtime_usec > request.if_modified_since_usec);
2920 SelectedJournalFiles {
2921 files: selected,
2922 files_are_newer,
2923 }
2924}
2925
2926fn remaining_files_cannot_affect_data_page(
2927 combined: &CombinedResult,
2928 request: &NetdataRequest,
2929 files: &[SelectedJournalFile],
2930 next_file_index: usize,
2931) -> bool {
2932 let Some(next) = files.get(next_file_index) else {
2933 return true;
2934 };
2935 let slack = next.order.journal_vs_realtime_delta_usec;
2936 match request.direction {
2937 Direction::Backward => {
2938 let Some(oldest_retained) = combined.rows.iter().map(|row| row.row.realtime_usec).min()
2939 else {
2940 return false;
2941 };
2942 next.order.msg_last_realtime_usec < oldest_retained.saturating_sub(slack)
2943 }
2944 Direction::Forward => {
2945 let Some(newest_retained) = combined.rows.iter().map(|row| row.row.realtime_usec).max()
2946 else {
2947 return false;
2948 };
2949 next.order.msg_first_realtime_usec > newest_retained.saturating_add(slack)
2950 }
2951 }
2952}
2953
2954fn journal_file_order_may_overlap_request(
2955 info: &JournalFileOrderInfo,
2956 request: &NetdataRequest,
2957) -> bool {
2958 if info.msg_last_realtime_usec == 0 {
2959 return true;
2960 }
2961
2962 let first = info
2963 .msg_first_realtime_usec
2964 .saturating_sub(NETDATA_JOURNAL_VS_REALTIME_DELTA_MAX_USEC);
2965 let last = info
2966 .msg_last_realtime_usec
2967 .saturating_add(NETDATA_JOURNAL_VS_REALTIME_DELTA_MAX_USEC);
2968
2969 if request
2970 .after_realtime_usec
2971 .is_some_and(|after| last < after)
2972 {
2973 return false;
2974 }
2975 if request
2976 .before_realtime_usec
2977 .is_some_and(|before| first > before)
2978 {
2979 return false;
2980 }
2981
2982 true
2983}
2984
2985fn collect_boot_first_realtime(
2986 paths: &[PathBuf],
2987 reader_options: ReaderOptions,
2988 needed_boot_ids: &BTreeSet<Vec<u8>>,
2989) -> BTreeMap<Vec<u8>, u64> {
2990 let mut out = BTreeMap::new();
2991 if needed_boot_ids.is_empty() {
2992 return out;
2993 }
2994 for path in paths {
2995 let Ok(mut reader) = FileReader::open_with_options(path, reader_options) else {
2996 continue;
2997 };
2998 let Ok(boot_ids) = reader.query_unique("_BOOT_ID") else {
2999 continue;
3000 };
3001 for boot_id in boot_ids {
3002 if !needed_boot_ids.contains(&boot_id) {
3003 continue;
3004 }
3005 let mut match_payload = b"_BOOT_ID=".to_vec();
3006 match_payload.extend_from_slice(&boot_id);
3007 reader.flush_matches();
3008 reader.add_match(&match_payload);
3009 reader.seek_head();
3010 if !reader.next().unwrap_or(false) {
3011 continue;
3012 }
3013 if let Ok(realtime) = reader.get_realtime_usec() {
3014 record_boot_first_realtime(&mut out, boot_id, realtime);
3015 }
3016 }
3017 reader.flush_matches();
3018 }
3019 out
3020}
3021
3022fn response_boot_ids(
3023 column_order: &[String],
3024 rows: &[LocatedRow],
3025 facets: &BTreeMap<Vec<u8>, BTreeMap<Vec<u8>, u64>>,
3026 histogram: Option<&ExplorerHistogram>,
3027) -> BTreeSet<Vec<u8>> {
3028 let mut boot_ids = BTreeSet::new();
3029 let row_needs_boot_id = column_order.iter().any(|field| field == "_BOOT_ID");
3030 if row_needs_boot_id {
3031 for row in rows {
3032 if let Some(values) = row_fields(row).get("_BOOT_ID") {
3033 boot_ids.extend(values.iter().cloned());
3034 }
3035 }
3036 }
3037 if let Some(values) = facets.get(b"_BOOT_ID".as_slice()) {
3038 boot_ids.extend(
3039 values
3040 .keys()
3041 .filter(|value| !value.is_empty() && value.as_slice() != b"-")
3042 .cloned(),
3043 );
3044 }
3045 if let Some(histogram) = histogram.filter(|histogram| histogram.field == b"_BOOT_ID") {
3046 for bucket in &histogram.buckets {
3047 boot_ids.extend(
3048 bucket
3049 .values
3050 .keys()
3051 .filter(|value| !value.is_empty() && value.as_slice() != b"-")
3052 .cloned(),
3053 );
3054 }
3055 }
3056 boot_ids
3057}
3058
3059fn record_boot_first_realtime(
3060 target: &mut BTreeMap<Vec<u8>, u64>,
3061 boot_id: Vec<u8>,
3062 realtime_usec: u64,
3063) {
3064 let existing = target.entry(boot_id).or_insert(realtime_usec);
3065 if realtime_usec < *existing {
3066 *existing = realtime_usec;
3067 }
3068}
3069
3070fn row_fields(row: &LocatedRow) -> BTreeMap<String, Vec<Vec<u8>>> {
3071 let mut fields = BTreeMap::new();
3072 for payload in &row.row.payloads {
3073 let Some((field, value)) = split_payload(payload) else {
3074 continue;
3075 };
3076 fields
3077 .entry(String::from_utf8_lossy(field).into_owned())
3078 .or_insert_with(Vec::new)
3079 .push(value.to_vec());
3080 }
3081 fields.insert(
3082 "ND_JOURNAL_FILE".to_string(),
3083 vec![row.file_path.display().to_string().into_bytes()],
3084 );
3085 if !fields.contains_key("ND_JOURNAL_PROCESS") {
3086 let process = dynamic_process_name(&fields);
3087 if !process.is_empty() {
3088 fields.insert("ND_JOURNAL_PROCESS".to_string(), vec![process.into_bytes()]);
3089 }
3090 }
3091 fields
3092}
3093
3094fn dynamic_process_name(fields: &BTreeMap<String, Vec<Vec<u8>>>) -> String {
3095 let base = first_value(fields, "CONTAINER_NAME")
3096 .or_else(|| first_value(fields, "SYSLOG_IDENTIFIER"))
3097 .or_else(|| first_value(fields, "_COMM"))
3098 .map(|value| String::from_utf8_lossy(value).into_owned())
3099 .unwrap_or_default();
3100 if base.is_empty() {
3101 return "-".to_string();
3102 }
3103 match first_value(fields, "_PID") {
3104 Some(pid) if !pid.is_empty() => {
3105 let pid = String::from_utf8_lossy(pid);
3106 format!("{base}[{pid}]")
3107 }
3108 Some(_) => base,
3109 None => format!("{base}[-]"),
3110 }
3111}
3112
3113fn first_value<'a>(fields: &'a BTreeMap<String, Vec<Vec<u8>>>, field: &str) -> Option<&'a [u8]> {
3114 fields
3115 .get(field)
3116 .and_then(|values| values.first())
3117 .map(Vec::as_slice)
3118}
3119
3120fn split_payload(payload: &[u8]) -> Option<(&[u8], &[u8])> {
3121 let split = payload.iter().position(|byte| *byte == b'=')?;
3122 Some((&payload[..split], &payload[split + 1..]))
3123}
3124
3125fn sort_rows_by_raw_realtime(rows: &mut [LocatedRow], direction: Direction) {
3126 rows.sort_by(|left, right| {
3127 let left_usec = left.row.realtime_usec;
3128 let right_usec = right.row.realtime_usec;
3129 let timestamp = match direction {
3130 Direction::Forward => left_usec.cmp(&right_usec),
3131 Direction::Backward => right_usec.cmp(&left_usec),
3132 };
3133 timestamp
3134 .then_with(|| left.file_path.cmp(&right.file_path))
3135 .then_with(|| left.row.cursor.cmp(&right.row.cursor))
3136 });
3137}
3138
3139fn retain_boundary_group(rows: &mut Vec<LocatedRow>, direction: Direction, limit: usize) {
3140 if limit == 0 || rows.is_empty() {
3141 if limit == 0 {
3142 rows.clear();
3143 }
3144 return;
3145 }
3146 if rows.len() <= limit {
3147 return;
3148 }
3149 let boundary_index = limit - 1;
3150 let boundary_timestamp = rows[boundary_index].row.realtime_usec;
3151 let keep_count = match direction {
3152 Direction::Backward => rows
3153 .iter()
3154 .position(|row| row.row.realtime_usec < boundary_timestamp)
3155 .unwrap_or(rows.len()),
3156 Direction::Forward => rows
3157 .iter()
3158 .rposition(|row| row.row.realtime_usec <= boundary_timestamp)
3159 .map(|index| index + 1)
3160 .unwrap_or(0),
3161 };
3162 rows.truncate(keep_count);
3163}
3164
3165fn make_row_timestamps_unique_same_file(rows: &mut [LocatedRow], direction: Direction) {
3166 let mut last_from = 0u64;
3167 let mut last_to = 0u64;
3168 let mut last_file: Option<&std::path::Path> = None;
3169 let mut initialized = false;
3170 for row in rows.iter_mut() {
3171 let timestamp = row.row.realtime_usec;
3172 let same_file = last_file == Some(row.file_path.as_path());
3173 if initialized && same_file && timestamp >= last_from && timestamp <= last_to {
3174 match direction {
3175 Direction::Backward => {
3176 last_from = last_from.saturating_sub(1);
3177 row.row.realtime_usec = last_from;
3178 }
3179 Direction::Forward => {
3180 last_to = last_to.saturating_add(1);
3181 row.row.realtime_usec = last_to;
3182 }
3183 }
3184 continue;
3185 }
3186 last_from = timestamp;
3187 last_to = timestamp;
3188 last_file = Some(row.file_path.as_path());
3189 initialized = true;
3190 }
3191}
3192
3193fn column_metadata(key: &str, index: usize) -> Value {
3194 let (visible, filter, full_width) = match key {
3195 "timestamp" => (true, "range", false),
3196 "rowOptions" => (false, "none", false),
3197 "_HOSTNAME" => (true, "facet", false),
3198 "ND_JOURNAL_PROCESS" | "MESSAGE" => (true, "none", key == "MESSAGE"),
3199 "ND_JOURNAL_FILE" | "_SOURCE_REALTIME_TIMESTAMP" => (false, "none", false),
3200 _ if systemd_column_is_facet(key) => (false, "facet", false),
3201 _ => (false, "none", false),
3202 };
3203 let column_type = if key == "timestamp" {
3204 "timestamp"
3205 } else if key == "rowOptions" {
3206 "none"
3207 } else {
3208 "string"
3209 };
3210 let visualization = if key == "rowOptions" {
3211 "rowOptions"
3212 } else {
3213 "value"
3214 };
3215 let mut metadata = json!({
3216 "index": index,
3217 "unique_key": key == "timestamp",
3218 "name": if key == "timestamp" { "Timestamp" } else { key },
3219 "visible": visible,
3220 "type": column_type,
3221 "visualization": visualization,
3222 "value_options": {
3223 "transform": if key == "timestamp" { "datetime_usec" } else { "none" },
3224 "decimal_points": 0,
3225 "default_value": if key == "timestamp" || key == "rowOptions" {
3226 Value::Null
3227 } else {
3228 Value::String("-".to_string())
3229 },
3230 },
3231 "sort": "ascending",
3232 "sortable": false,
3233 "sticky": false,
3234 "summary": "count",
3235 "filter": filter,
3236 "full_width": full_width,
3237 "wrap": key != "rowOptions",
3238 "default_expanded_filter": matches!(key, "PRIORITY" | "SYSLOG_FACILITY" | "MESSAGE_ID"),
3239 });
3240 if key == "rowOptions" {
3241 if let Some(object) = metadata.as_object_mut() {
3242 object.insert("dummy".to_string(), Value::Bool(true));
3243 }
3244 }
3245 metadata
3246}
3247
3248fn systemd_column_is_facet(key: &str) -> bool {
3249 if key == "MESSAGE_ID" {
3250 return true;
3251 }
3252 if key.contains("MESSAGE") || key.contains("TIMESTAMP") || key.starts_with("__") {
3253 return false;
3254 }
3255 true
3256}
3257
3258fn sort_facet_options(field: &str, options: &mut [Value]) {
3259 options.sort_by(|left, right| {
3260 let left_id = left.get("id").and_then(Value::as_str).unwrap_or_default();
3261 let right_id = right.get("id").and_then(Value::as_str).unwrap_or_default();
3262 if field == "PRIORITY" {
3263 return parse_priority(left_id).cmp(&parse_priority(right_id));
3264 }
3265 let left_count = left
3266 .get("count")
3267 .and_then(Value::as_u64)
3268 .unwrap_or_default();
3269 let right_count = right
3270 .get("count")
3271 .and_then(Value::as_u64)
3272 .unwrap_or_default();
3273 right_count
3274 .cmp(&left_count)
3275 .then_with(|| left_id.cmp(right_id))
3276 });
3277}
3278
3279fn parse_fts_query_patterns(query: &str) -> (Vec<ExplorerFtsPattern>, Vec<Vec<u8>>, Vec<Vec<u8>>) {
3280 let bytes = query.as_bytes();
3281 let mut index = 0usize;
3282 let mut terms = Vec::new();
3283 let mut positives = Vec::new();
3284 let mut negatives = Vec::new();
3285
3286 while let Some((pattern, negative)) = next_fts_pattern(bytes, &mut index) {
3287 push_fts_pattern(
3288 pattern,
3289 negative,
3290 &mut terms,
3291 &mut positives,
3292 &mut negatives,
3293 );
3294 }
3295
3296 (terms, positives, negatives)
3297}
3298
3299fn next_fts_pattern(bytes: &[u8], index: &mut usize) -> Option<(Vec<u8>, bool)> {
3300 while *index < bytes.len() {
3301 skip_fts_separators(bytes, index);
3302 let negative = consume_fts_negative_marker(bytes, index);
3303 let pattern = read_fts_pattern(bytes, index);
3304 if !pattern.is_empty() {
3305 return Some((pattern, negative));
3306 }
3307 }
3308 None
3309}
3310
3311fn skip_fts_separators(bytes: &[u8], index: &mut usize) {
3312 while *index < bytes.len() && bytes[*index] == b'|' {
3313 *index += 1;
3314 }
3315}
3316
3317fn consume_fts_negative_marker(bytes: &[u8], index: &mut usize) -> bool {
3318 if bytes.get(*index) == Some(&b'!') {
3319 *index += 1;
3320 true
3321 } else {
3322 false
3323 }
3324}
3325
3326fn read_fts_pattern(bytes: &[u8], index: &mut usize) -> Vec<u8> {
3327 let mut pattern = Vec::new();
3328 let mut escaped = false;
3329 while *index < bytes.len() {
3330 let byte = bytes[*index];
3331 *index += 1;
3332 if byte == b'\\' && !escaped {
3333 escaped = true;
3334 continue;
3335 }
3336 if byte == b'|' && !escaped {
3337 break;
3338 }
3339 pattern.push(byte);
3340 escaped = false;
3341 }
3342 pattern
3343}
3344
3345fn push_fts_pattern(
3346 pattern: Vec<u8>,
3347 negative: bool,
3348 terms: &mut Vec<ExplorerFtsPattern>,
3349 positives: &mut Vec<Vec<u8>>,
3350 negatives: &mut Vec<Vec<u8>>,
3351) {
3352 terms.push(ExplorerFtsPattern::substring(pattern.clone(), negative));
3353 if negative {
3354 negatives.push(pattern);
3355 } else {
3356 positives.push(pattern);
3357 }
3358}
3359
3360fn parse_filters(value: Option<&Value>) -> Vec<ExplorerFilter> {
3361 let Some(Value::Object(selections)) = value else {
3362 return Vec::new();
3363 };
3364 let mut filters = Vec::new();
3365 for (field, values) in selections {
3366 if matches!(field.as_str(), "query" | "source" | "__logs_sources") {
3367 continue;
3368 }
3369 let Some(values) = parse_string_array(Some(values)) else {
3370 continue;
3371 };
3372 filters.push(ExplorerFilter::new(
3373 field.as_bytes().to_vec(),
3374 values
3375 .into_iter()
3376 .map(|value| normalize_filter_value(field, &value)),
3377 ));
3378 }
3379 filters
3380}
3381
3382#[derive(Debug, Clone)]
3383struct SourceSelection {
3384 source_type: u64,
3385 exact_sources: Vec<String>,
3386}
3387
3388fn parse_source_selection(value: Option<&Value>) -> SourceSelection {
3389 let mut selection = SourceSelection {
3390 source_type: SOURCE_TYPE_ALL,
3391 exact_sources: Vec::new(),
3392 };
3393 let Some(Value::Object(selections)) = value else {
3394 return selection;
3395 };
3396 let Some(values) = parse_string_array(selections.get("__logs_sources")) else {
3397 return selection;
3398 };
3399 selection.source_type = 0;
3400 for value in values {
3401 match source_type_for_name(&value) {
3402 Some(source_type) => selection.source_type |= source_type,
3403 None => selection.exact_sources.push(value),
3404 }
3405 }
3406 selection
3407}
3408
3409fn source_type_for_name(value: &str) -> Option<u64> {
3410 match value {
3411 "all" => Some(SOURCE_TYPE_ALL),
3412 "all-local-logs" => Some(SOURCE_TYPE_LOCAL_ALL),
3413 "all-remote-systems" => Some(SOURCE_TYPE_REMOTE_ALL),
3414 "all-local-system-logs" => Some(SOURCE_TYPE_LOCAL_SYSTEM),
3415 "all-local-user-logs" => Some(SOURCE_TYPE_LOCAL_USER),
3416 "all-local-namespaces" => Some(SOURCE_TYPE_LOCAL_NAMESPACE),
3417 "all-uncategorized" => Some(SOURCE_TYPE_LOCAL_OTHER),
3418 _ => None,
3419 }
3420}
3421
3422fn journal_file_source_type(path: &Path) -> u64 {
3423 let text = path.to_string_lossy();
3424 let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
3425 return SOURCE_TYPE_ALL | SOURCE_TYPE_LOCAL_ALL | SOURCE_TYPE_LOCAL_OTHER;
3426 };
3427 if text.contains("/remote/") {
3428 return SOURCE_TYPE_ALL | SOURCE_TYPE_REMOTE_ALL;
3429 }
3430 if local_namespace_source_name(path).is_some() {
3431 return SOURCE_TYPE_ALL | SOURCE_TYPE_LOCAL_ALL | SOURCE_TYPE_LOCAL_NAMESPACE;
3432 }
3433 if name.starts_with("system") {
3434 return SOURCE_TYPE_ALL | SOURCE_TYPE_LOCAL_ALL | SOURCE_TYPE_LOCAL_SYSTEM;
3435 }
3436 if name.starts_with("user") {
3437 return SOURCE_TYPE_ALL | SOURCE_TYPE_LOCAL_ALL | SOURCE_TYPE_LOCAL_USER;
3438 }
3439 SOURCE_TYPE_ALL | SOURCE_TYPE_LOCAL_ALL | SOURCE_TYPE_LOCAL_OTHER
3440}
3441
3442fn local_namespace_source_name(path: &Path) -> Option<String> {
3443 let parent = path.parent()?.file_name()?.to_str()?;
3444 let (_, namespace) = parent.rsplit_once('.')?;
3445 (!namespace.is_empty()).then(|| format!("namespace-{namespace}"))
3446}
3447
3448fn journal_file_exact_source_name(path: &Path) -> Option<String> {
3449 let text = path.to_string_lossy();
3450 if text.contains("/remote/") {
3451 let name = path.file_name()?.to_str()?;
3452 let source = name
3453 .split_once('@')
3454 .map(|(prefix, _)| prefix)
3455 .unwrap_or_else(|| {
3456 name.strip_suffix(".journal~.zst")
3457 .or_else(|| name.strip_suffix(".journal.zst"))
3458 .or_else(|| name.strip_suffix(".journal~"))
3459 .or_else(|| name.strip_suffix(".journal"))
3460 .unwrap_or(name)
3461 });
3462 return source.starts_with("remote-").then(|| source.to_string());
3463 }
3464 local_namespace_source_name(path)
3465}
3466
3467fn normalize_filter_value(field: &str, value: &str) -> Vec<u8> {
3468 if field == "PRIORITY" {
3469 if let Some(priority) = priority_name_to_number(value) {
3470 return priority.as_bytes().to_vec();
3471 }
3472 }
3473 value.as_bytes().to_vec()
3474}
3475
3476fn parse_string_array(value: Option<&Value>) -> Option<Vec<String>> {
3477 let Value::Array(items) = value? else {
3478 return None;
3479 };
3480 Some(
3481 items
3482 .iter()
3483 .filter_map(Value::as_str)
3484 .map(ToOwned::to_owned)
3485 .collect(),
3486 )
3487}
3488
3489fn request_direction(object: &Map<String, Value>) -> Direction {
3490 match get_str(object, "direction").unwrap_or("backward") {
3491 "forward" | "forwards" | "next" => Direction::Forward,
3492 _ => Direction::Backward,
3493 }
3494}
3495
3496fn request_delta(data_only: bool, object: &Map<String, Value>) -> bool {
3497 data_only && get_bool(object, "delta").unwrap_or(false)
3498}
3499
3500fn request_tail(data_only: bool, if_modified_since_usec: u64, object: &Map<String, Value>) -> bool {
3501 data_only && if_modified_since_usec != 0 && get_bool(object, "tail").unwrap_or(false)
3502}
3503
3504fn tail_after_realtime_bound(
3505 after_realtime_usec: Option<u64>,
3506 anchor: ExplorerAnchor,
3507) -> Option<u64> {
3508 let ExplorerAnchor::Realtime(anchor) = anchor else {
3509 return after_realtime_usec;
3510 };
3511 let tail_after = anchor.saturating_add(1);
3512 Some(
3513 after_realtime_usec
3514 .map(|after| after.max(tail_after))
3515 .unwrap_or(tail_after),
3516 )
3517}
3518
3519fn before_realtime_bound_excluding_anchor(
3520 before_realtime_usec: Option<u64>,
3521 anchor: ExplorerAnchor,
3522) -> Option<u64> {
3523 let ExplorerAnchor::Realtime(anchor) = anchor else {
3524 return before_realtime_usec;
3525 };
3526 let before_anchor = anchor.saturating_sub(1);
3527 Some(
3528 before_realtime_usec
3529 .map(|before| before.min(before_anchor))
3530 .unwrap_or(before_anchor),
3531 )
3532}
3533
3534fn request_anchor_and_direction(
3535 object: &Map<String, Value>,
3536 tail: bool,
3537 direction: Direction,
3538 after_realtime_usec: Option<u64>,
3539 before_realtime_usec: Option<u64>,
3540) -> (ExplorerAnchor, Direction) {
3541 let anchor = get_u64(object, "anchor")
3542 .filter(|value| *value != 0)
3543 .map(normalize_timestamp_to_usec)
3544 .map(ExplorerAnchor::Realtime)
3545 .unwrap_or(ExplorerAnchor::Auto);
3546 if tail && matches!(anchor, ExplorerAnchor::Realtime(_)) {
3547 return (anchor, Direction::Backward);
3548 }
3549 if anchor_outside_window(anchor, after_realtime_usec, before_realtime_usec) {
3550 (ExplorerAnchor::Auto, Direction::Backward)
3551 } else {
3552 (anchor, direction)
3553 }
3554}
3555
3556fn anchor_outside_window(
3557 anchor: ExplorerAnchor,
3558 after_realtime_usec: Option<u64>,
3559 before_realtime_usec: Option<u64>,
3560) -> bool {
3561 let ExplorerAnchor::Realtime(anchor_usec) = anchor else {
3562 return false;
3563 };
3564 after_realtime_usec.is_some_and(|after| anchor_usec < after)
3565 || before_realtime_usec.is_some_and(|before| anchor_usec > before)
3566}
3567
3568fn request_limit(object: &Map<String, Value>) -> usize {
3569 get_u64(object, "last")
3570 .filter(|value| *value != 0)
3571 .map(|value| value as usize)
3572 .unwrap_or(DEFAULT_ITEMS_TO_RETURN)
3573}
3574
3575fn request_facets(
3576 requested_facets: &Option<Vec<String>>,
3577 config: &NetdataFunctionConfig,
3578) -> Vec<Vec<u8>> {
3579 requested_facets
3580 .clone()
3581 .unwrap_or_else(|| config.default_facets.clone())
3582 .into_iter()
3583 .map(Vec::from)
3584 .collect()
3585}
3586
3587fn request_histogram(object: &Map<String, Value>) -> Option<String> {
3588 get_str(object, "histogram")
3589 .filter(|histogram| !histogram.is_empty())
3590 .map(ToOwned::to_owned)
3591}
3592
3593fn request_histogram_or_default(
3594 requested_histogram: &Option<String>,
3595 config: &NetdataFunctionConfig,
3596) -> Option<String> {
3597 requested_histogram
3598 .clone()
3599 .or_else(|| config.default_histogram.clone())
3600}
3601
3602fn request_query(object: &Map<String, Value>) -> Option<String> {
3603 get_str(object, "query")
3604 .filter(|query| !query.is_empty())
3605 .map(ToOwned::to_owned)
3606}
3607
3608fn get_bool(object: &Map<String, Value>, key: &str) -> Option<bool> {
3609 object.get(key).and_then(Value::as_bool)
3610}
3611
3612fn get_i64(object: &Map<String, Value>, key: &str) -> Option<i64> {
3613 object.get(key).and_then(Value::as_i64)
3614}
3615
3616fn get_u64(object: &Map<String, Value>, key: &str) -> Option<u64> {
3617 object.get(key).and_then(Value::as_u64)
3618}
3619
3620fn get_str<'a>(object: &'a Map<String, Value>, key: &str) -> Option<&'a str> {
3621 object.get(key).and_then(Value::as_str)
3622}
3623
3624fn normalize_time_window(
3625 now_seconds: i64,
3626 after: Option<i64>,
3627 before: Option<i64>,
3628) -> (Option<u64>, Option<u64>) {
3629 let mut after = after.unwrap_or(0);
3630 let mut before = before.unwrap_or(0);
3631
3632 if after == 0 && before == 0 {
3633 before = now_seconds;
3634 after = before.saturating_sub(DEFAULT_TIME_WINDOW_SECONDS);
3635 } else {
3636 (after, before) = relative_window_to_absolute(now_seconds, after, before);
3637 }
3638
3639 if after > before {
3640 std::mem::swap(&mut after, &mut before);
3641 }
3642 if after == before {
3643 after = before.saturating_sub(DEFAULT_TIME_WINDOW_SECONDS);
3644 }
3645
3646 (
3647 Some(normalize_timestamp_to_usec_with_rounding(
3648 after.max(0) as u64,
3649 false,
3650 )),
3651 Some(normalize_timestamp_to_usec_with_rounding(
3652 before.max(0) as u64,
3653 true,
3654 )),
3655 )
3656}
3657
3658fn relative_window_to_absolute(now_seconds: i64, after: i64, before: i64) -> (i64, i64) {
3659 let mut after = after;
3660 let mut before = before;
3661
3662 if before.unsigned_abs() <= API_RELATIVE_TIME_MAX_SECONDS as u64 {
3663 if before > 0 {
3664 before = -before;
3665 }
3666 before = now_seconds.saturating_add(before);
3667 }
3668
3669 if after.unsigned_abs() <= API_RELATIVE_TIME_MAX_SECONDS as u64 {
3670 if after > 0 {
3671 after = -after;
3672 }
3673 if after == 0 {
3674 after = -NETDATA_MISSING_AFTER_RELATIVE_SECONDS;
3675 }
3676 after = before.saturating_add(after).saturating_add(1);
3677 }
3678
3679 if after > before {
3680 std::mem::swap(&mut after, &mut before);
3681 }
3682
3683 if before > now_seconds {
3684 let delta = before.saturating_sub(now_seconds);
3685 before = before.saturating_sub(delta);
3686 after = after.saturating_sub(delta);
3687 }
3688
3689 (after, before)
3690}
3691
3692struct RequestEchoInput<'a> {
3693 info: bool,
3694 after_realtime_usec: Option<u64>,
3695 before_realtime_usec: Option<u64>,
3696 if_modified_since_usec: u64,
3697 anchor: ExplorerAnchor,
3698 direction: Direction,
3699 limit: usize,
3700 data_only: bool,
3701 delta: bool,
3702 tail: bool,
3703 sampling: u64,
3704 source_type: u64,
3705 requested_facets: Option<&'a [String]>,
3706 selections: Option<&'a Value>,
3707 histogram: Option<&'a str>,
3708 query: Option<&'a str>,
3709}
3710
3711fn normalized_request_echo(input: &RequestEchoInput<'_>) -> Value {
3712 let anchor_usec = match input.anchor {
3713 ExplorerAnchor::Realtime(usec) => usec,
3714 ExplorerAnchor::Auto | ExplorerAnchor::Head | ExplorerAnchor::Tail => 0,
3715 };
3716 let mut out = json!({
3717 "info": input.info,
3718 "slice": true,
3722 "data_only": input.data_only,
3723 "delta": input.delta,
3724 "tail": input.tail,
3725 "sampling": input.sampling,
3726 "source_type": input.source_type,
3727 "after": input.after_realtime_usec.unwrap_or(0) / 1_000_000,
3728 "before": input.before_realtime_usec.unwrap_or(0) / 1_000_000,
3729 "if_modified_since": input.if_modified_since_usec,
3730 "anchor": anchor_usec,
3731 "direction": match input.direction {
3732 Direction::Forward => "forward",
3733 Direction::Backward => "backward",
3734 },
3735 "last": input.limit,
3736 "query": input.query,
3737 "histogram": input.histogram,
3738 });
3739 if let Some(facets) = input.requested_facets {
3740 if let Some(object) = out.as_object_mut() {
3741 object.insert(
3742 "facets".to_string(),
3743 facets
3744 .iter()
3745 .map(|field| Value::String(field.clone()))
3746 .collect(),
3747 );
3748 }
3749 }
3750 if let Some(Value::Object(selections)) = input.selections {
3751 let mut selections = selections.clone();
3752 if let Some(Value::Array(sources)) = selections.get_mut("__logs_sources") {
3753 for source in sources {
3754 *source = Value::Null;
3755 }
3756 }
3757 if let Some(object) = out.as_object_mut() {
3758 object.insert("selections".to_string(), Value::Object(selections));
3759 }
3760 }
3761 out
3762}
3763
3764fn normalize_timestamp_to_usec(value: u64) -> u64 {
3765 normalize_timestamp_to_usec_with_rounding(value, false)
3766}
3767
3768fn normalize_timestamp_to_usec_with_rounding(value: u64, end_of_second: bool) -> u64 {
3769 if value >= 1_000_000_000_000 {
3770 value
3771 } else if end_of_second {
3772 value.saturating_mul(1_000_000).saturating_add(999_999)
3773 } else {
3774 value.saturating_mul(1_000_000)
3775 }
3776}
3777
3778fn unix_now_seconds() -> i64 {
3779 SystemTime::now()
3780 .duration_since(UNIX_EPOCH)
3781 .map(|duration| duration.as_secs() as i64)
3782 .unwrap_or_default()
3783}
3784
3785fn human_binary_size(bytes: u64) -> String {
3786 const UNITS: &[&str] = &["B", "KiB", "MiB", "GiB", "TiB"];
3787 let mut value = bytes as f64;
3788 let mut unit = 0usize;
3789 while value >= 1024.0 && unit + 1 < UNITS.len() {
3790 value /= 1024.0;
3791 unit += 1;
3792 }
3793 if unit == 0 {
3794 format!("{}{}", bytes, UNITS[unit])
3795 } else if value.fract() == 0.0 {
3796 format!("{value:.0}{}", UNITS[unit])
3797 } else {
3798 let mut formatted = format!("{value:.2}");
3799 while formatted.contains('.') && formatted.ends_with('0') {
3800 formatted.pop();
3801 }
3802 if formatted.ends_with('.') {
3803 formatted.pop();
3804 }
3805 format!("{formatted}{}", UNITS[unit])
3806 }
3807}
3808
3809fn human_duration_seconds(seconds: u64) -> String {
3810 let years = seconds / (365 * 86_400);
3811 let seconds = seconds % (365 * 86_400);
3812 let months = seconds / (30 * 86_400);
3813 let seconds = seconds % (30 * 86_400);
3814 let days = seconds / 86_400;
3815 let seconds = seconds % 86_400;
3816 let hours = seconds / 3600;
3817 let minutes = (seconds % 3600) / 60;
3818 let seconds = seconds % 60;
3819 let mut parts = Vec::new();
3820 if years != 0 {
3821 parts.push(format!("{years}y"));
3822 }
3823 if months != 0 {
3824 parts.push(format!("{months}mo"));
3825 }
3826 if days != 0 {
3827 parts.push(format!("{days}d"));
3828 }
3829 if hours != 0 {
3830 parts.push(format!("{hours}h"));
3831 }
3832 if minutes != 0 {
3833 parts.push(format!("{minutes}m"));
3834 }
3835 if seconds != 0 || parts.is_empty() {
3836 parts.push(format!("{seconds}s"));
3837 }
3838 parts.join(" ")
3839}
3840
3841#[derive(Debug, Default)]
3842struct JournalFileCollection {
3843 files: Vec<PathBuf>,
3844 skipped: u64,
3845 errors: Vec<String>,
3846}
3847
3848fn collect_journal_files(path: &Path) -> Result<JournalFileCollection> {
3849 if !path.is_dir() {
3850 return Err(SdkError::InvalidPath(format!(
3851 "not a directory: {}",
3852 path.display()
3853 )));
3854 }
3855 let mut collection = JournalFileCollection::default();
3856 let mut pending = VecDeque::from([(path.to_path_buf(), 0usize)]);
3857 let mut visited = HashSet::new();
3858 while let Some((directory, depth)) = pending.pop_front() {
3859 let visited_key = std::fs::canonicalize(&directory).unwrap_or_else(|_| directory.clone());
3860 if visited.contains(&visited_key) {
3861 continue;
3862 }
3863 if visited.len() >= NETDATA_MAX_DIRECTORY_SCAN_COUNT {
3864 collection.skipped = collection.skipped.saturating_add(1);
3865 collection.errors.push(format!(
3866 "{}: directory scan limit reached",
3867 directory.display()
3868 ));
3869 continue;
3870 }
3871 visited.insert(visited_key);
3872 let entries = match std::fs::read_dir(&directory) {
3873 Ok(entries) => entries,
3874 Err(err) if directory == path => return Err(err.into()),
3875 Err(err) => {
3876 collection.skipped = collection.skipped.saturating_add(1);
3877 collection
3878 .errors
3879 .push(format!("{}: {err}", directory.display()));
3880 continue;
3881 }
3882 };
3883 for entry in entries.flatten() {
3884 let entry_path = entry.path();
3885 if entry_path.is_file() && is_journal_file_name(&entry_path) {
3886 collection.files.push(entry_path);
3887 } else if depth < NETDATA_MAX_DIRECTORY_SCAN_DEPTH && entry_path.is_dir() {
3888 pending.push_back((entry_path, depth + 1));
3889 }
3890 }
3891 }
3892 collection.files.sort();
3893 dedup_journal_files_by_canonical_path(&mut collection.files);
3894 Ok(collection)
3895}
3896
3897fn dedup_journal_files_by_canonical_path(files: &mut Vec<PathBuf>) {
3898 let mut seen = HashSet::new();
3899 files.retain(|path| {
3900 let key = std::fs::canonicalize(path).unwrap_or_else(|_| path.clone());
3901 seen.insert(key)
3902 });
3903}
3904
3905#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3906struct JournalFileOrderInfo {
3907 msg_first_realtime_usec: u64,
3908 msg_last_realtime_usec: u64,
3909 file_last_modified_usec: u64,
3910 journal_vs_realtime_delta_usec: u64,
3911}
3912
3913fn journal_file_order_info(
3914 path: &Path,
3915 reader_options: ReaderOptions,
3916 metadata: Option<&NetdataJournalFileMetadata>,
3917) -> JournalFileOrderInfo {
3918 let file_last_modified_usec = std::fs::metadata(path)
3919 .ok()
3920 .and_then(|metadata| metadata.modified().ok())
3921 .and_then(|modified| modified.duration_since(UNIX_EPOCH).ok())
3922 .map(|duration| duration.as_micros().min(u128::from(u64::MAX)) as u64)
3923 .unwrap_or_default();
3924 let file_last_modified_usec = metadata
3925 .and_then(|metadata| metadata.file_last_modified_usec)
3926 .unwrap_or(file_last_modified_usec);
3927 let journal_vs_realtime_delta_usec = metadata
3928 .and_then(|metadata| metadata.journal_vs_realtime_delta_usec)
3929 .map(normalize_journal_vs_realtime_delta_usec)
3930 .unwrap_or(NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC);
3931
3932 let Ok(reader) = FileReader::open_with_options(path, reader_options) else {
3933 return JournalFileOrderInfo {
3934 msg_first_realtime_usec: 0,
3935 msg_last_realtime_usec: file_last_modified_usec,
3936 file_last_modified_usec,
3937 journal_vs_realtime_delta_usec,
3938 };
3939 };
3940 let header = reader.header();
3941 let msg_first_realtime_usec = metadata
3942 .and_then(|metadata| metadata.msg_first_realtime_usec)
3943 .unwrap_or(header.head_entry_realtime);
3944 let msg_last_realtime_usec = metadata
3945 .and_then(|metadata| metadata.msg_last_realtime_usec)
3946 .unwrap_or_else(|| {
3947 if header.tail_entry_realtime == 0 {
3948 file_last_modified_usec
3949 } else {
3950 header.tail_entry_realtime
3951 }
3952 });
3953 JournalFileOrderInfo {
3954 msg_first_realtime_usec,
3955 msg_last_realtime_usec,
3956 file_last_modified_usec,
3957 journal_vs_realtime_delta_usec,
3958 }
3959}
3960
3961fn compare_journal_file_order(
3962 left: &JournalFileOrderInfo,
3963 right: &JournalFileOrderInfo,
3964 direction: Direction,
3965) -> Ordering {
3966 let backward = right
3967 .msg_last_realtime_usec
3968 .cmp(&left.msg_last_realtime_usec)
3969 .then_with(|| {
3970 right
3971 .file_last_modified_usec
3972 .cmp(&left.file_last_modified_usec)
3973 })
3974 .then_with(|| {
3975 right
3976 .msg_first_realtime_usec
3977 .cmp(&left.msg_first_realtime_usec)
3978 });
3979 match direction {
3980 Direction::Backward => backward,
3981 Direction::Forward => backward.reverse(),
3982 }
3983}
3984
3985fn is_journal_file_name(path: &Path) -> bool {
3986 path.file_name()
3987 .and_then(|name| name.to_str())
3988 .is_some_and(|name| {
3989 name.ends_with(".journal")
3990 || name.ends_with(".journal~")
3991 || name.ends_with(".journal.zst")
3992 || name.ends_with(".journal~.zst")
3993 })
3994}
3995
3996fn push_unique_many(target: &mut Vec<String>, values: &[String]) {
3997 for value in values {
3998 push_unique(target, value);
3999 }
4000}
4001
4002fn string_fields(fields: &[Vec<u8>]) -> Vec<String> {
4003 fields
4004 .iter()
4005 .filter_map(|field| String::from_utf8(field.clone()).ok())
4006 .collect()
4007}
4008
4009fn push_unique(target: &mut Vec<String>, value: impl AsRef<str>) {
4010 let value = value.as_ref();
4011 if !target.iter().any(|existing| existing == value) {
4012 target.push(value.to_string());
4013 }
4014}
4015
4016fn netdata_reorder_key(value: &str) -> String {
4017 value
4018 .trim_start_matches(|character: char| character.is_ascii_punctuation())
4019 .to_ascii_lowercase()
4020}
4021
4022fn histogram_update_every_seconds(histogram: &ExplorerHistogram) -> u64 {
4023 histogram
4024 .buckets
4025 .first()
4026 .map(|bucket| {
4027 bucket
4028 .end_realtime_usec
4029 .saturating_sub(bucket.start_realtime_usec)
4030 .checked_div(1_000_000)
4031 .unwrap_or(1)
4032 .max(1)
4033 })
4034 .unwrap_or(1)
4035}
4036
4037enum TimestampPrecision {
4038 Seconds,
4039 Micros,
4040}
4041
4042fn format_realtime_usec(timestamp: u64, precision: TimestampPrecision) -> String {
4043 let seconds = (timestamp / 1_000_000) as i64;
4044 let micros = (timestamp % 1_000_000) as u32;
4045 DateTime::<Utc>::from_timestamp(seconds, micros.saturating_mul(1000))
4046 .map(|datetime| match precision {
4047 TimestampPrecision::Seconds => datetime.format("%Y-%m-%dT%H:%M:%SZ").to_string(),
4048 TimestampPrecision::Micros => datetime.format("%Y-%m-%dT%H:%M:%S%.6fZ").to_string(),
4049 })
4050 .unwrap_or_else(|| timestamp.to_string())
4051}
4052
4053fn priority_name(raw: &str) -> Option<&'static str> {
4054 match parse_priority(raw)? {
4055 0 => Some("panic"),
4056 1 => Some("alert"),
4057 2 => Some("critical"),
4058 3 => Some("error"),
4059 4 => Some("warning"),
4060 5 => Some("notice"),
4061 6 => Some("info"),
4062 7 => Some("debug"),
4063 _ => None,
4064 }
4065}
4066
4067fn priority_name_to_number(value: &str) -> Option<&'static str> {
4068 match value {
4069 "panic" | "emergency" | "emerg" => Some("0"),
4070 "alert" => Some("1"),
4071 "critical" | "crit" => Some("2"),
4072 "error" | "err" => Some("3"),
4073 "warning" | "warn" => Some("4"),
4074 "notice" => Some("5"),
4075 "info" => Some("6"),
4076 "debug" => Some("7"),
4077 _ => None,
4078 }
4079}
4080
4081fn parse_priority(raw: &str) -> Option<u8> {
4082 raw.parse::<u8>().ok()
4083}
4084
4085fn priority_to_row_severity(raw: &[u8]) -> &'static str {
4086 let raw = String::from_utf8_lossy(raw);
4087 match parse_priority(&raw) {
4088 Some(priority) if priority <= 3 => "critical",
4089 Some(4) => "warning",
4090 Some(5) => "notice",
4091 Some(priority) if priority >= 7 => "debug",
4092 _ => "normal",
4093 }
4094}
4095
4096fn syslog_facility_name(raw: &str) -> Option<&'static str> {
4097 match raw.parse::<u8>().ok()? {
4098 0 => Some("kern"),
4099 1 => Some("user"),
4100 2 => Some("mail"),
4101 3 => Some("daemon"),
4102 4 => Some("auth"),
4103 5 => Some("syslog"),
4104 6 => Some("lpr"),
4105 7 => Some("news"),
4106 8 => Some("uucp"),
4107 9 => Some("cron"),
4108 10 => Some("authpriv"),
4109 11 => Some("ftp"),
4110 16 => Some("local0"),
4111 17 => Some("local1"),
4112 18 => Some("local2"),
4113 19 => Some("local3"),
4114 20 => Some("local4"),
4115 21 => Some("local5"),
4116 22 => Some("local6"),
4117 23 => Some("local7"),
4118 _ => None,
4119 }
4120}
4121
4122const ERRNO_NAMES: &[(u32, &str)] = &[
4123 (1, "EPERM"),
4124 (2, "ENOENT"),
4125 (3, "ESRCH"),
4126 (4, "EINTR"),
4127 (5, "EIO"),
4128 (6, "ENXIO"),
4129 (7, "E2BIG"),
4130 (8, "ENOEXEC"),
4131 (9, "EBADF"),
4132 (10, "ECHILD"),
4133 (11, "EAGAIN"),
4134 (12, "ENOMEM"),
4135 (13, "EACCES"),
4136 (14, "EFAULT"),
4137 (15, "ENOTBLK"),
4138 (16, "EBUSY"),
4139 (17, "EEXIST"),
4140 (18, "EXDEV"),
4141 (19, "ENODEV"),
4142 (20, "ENOTDIR"),
4143 (21, "EISDIR"),
4144 (22, "EINVAL"),
4145 (23, "ENFILE"),
4146 (24, "EMFILE"),
4147 (25, "ENOTTY"),
4148 (26, "ETXTBSY"),
4149 (27, "EFBIG"),
4150 (28, "ENOSPC"),
4151 (29, "ESPIPE"),
4152 (30, "EROFS"),
4153 (31, "EMLINK"),
4154 (32, "EPIPE"),
4155 (33, "EDOM"),
4156 (34, "ERANGE"),
4157 (35, "EDEADLK"),
4158 (36, "ENAMETOOLONG"),
4159 (37, "ENOLCK"),
4160 (38, "ENOSYS"),
4161 (39, "ENOTEMPTY"),
4162 (40, "ELOOP"),
4163 (42, "ENOMSG"),
4164 (43, "EIDRM"),
4165 (44, "ECHRNG"),
4166 (45, "EL2NSYNC"),
4167 (46, "EL3HLT"),
4168 (47, "EL3RST"),
4169 (48, "ELNRNG"),
4170 (49, "EUNATCH"),
4171 (50, "ENOCSI"),
4172 (51, "EL2HLT"),
4173 (52, "EBADE"),
4174 (53, "EBADR"),
4175 (54, "EXFULL"),
4176 (55, "ENOANO"),
4177 (56, "EBADRQC"),
4178 (57, "EBADSLT"),
4179 (59, "EBFONT"),
4180 (60, "ENOSTR"),
4181 (61, "ENODATA"),
4182 (62, "ETIME"),
4183 (63, "ENOSR"),
4184 (64, "ENONET"),
4185 (65, "ENOPKG"),
4186 (66, "EREMOTE"),
4187 (67, "ENOLINK"),
4188 (68, "EADV"),
4189 (69, "ESRMNT"),
4190 (70, "ECOMM"),
4191 (71, "EPROTO"),
4192 (72, "EMULTIHOP"),
4193 (73, "EDOTDOT"),
4194 (74, "EBADMSG"),
4195 (75, "EOVERFLOW"),
4196 (76, "ENOTUNIQ"),
4197 (77, "EBADFD"),
4198 (78, "EREMCHG"),
4199 (79, "ELIBACC"),
4200 (80, "ELIBBAD"),
4201 (81, "ELIBSCN"),
4202 (82, "ELIBMAX"),
4203 (83, "ELIBEXEC"),
4204 (84, "EILSEQ"),
4205 (85, "ERESTART"),
4206 (86, "ESTRPIPE"),
4207 (87, "EUSERS"),
4208 (88, "ENOTSOCK"),
4209 (89, "EDESTADDRREQ"),
4210 (90, "EMSGSIZE"),
4211 (91, "EPROTOTYPE"),
4212 (92, "ENOPROTOOPT"),
4213 (93, "EPROTONOSUPPORT"),
4214 (94, "ESOCKTNOSUPPORT"),
4215 (95, "ENOTSUP"),
4216 (96, "EPFNOSUPPORT"),
4217 (97, "EAFNOSUPPORT"),
4218 (98, "EADDRINUSE"),
4219 (99, "EADDRNOTAVAIL"),
4220 (100, "ENETDOWN"),
4221 (101, "ENETUNREACH"),
4222 (102, "ENETRESET"),
4223 (103, "ECONNABORTED"),
4224 (104, "ECONNRESET"),
4225 (105, "ENOBUFS"),
4226 (106, "EISCONN"),
4227 (107, "ENOTCONN"),
4228 (108, "ESHUTDOWN"),
4229 (109, "ETOOMANYREFS"),
4230 (110, "ETIMEDOUT"),
4231 (111, "ECONNREFUSED"),
4232 (112, "EHOSTDOWN"),
4233 (113, "EHOSTUNREACH"),
4234 (114, "EALREADY"),
4235 (115, "EINPROGRESS"),
4236 (116, "ESTALE"),
4237 (117, "EUCLEAN"),
4238 (118, "ENOTNAM"),
4239 (119, "ENAVAIL"),
4240 (120, "EISNAM"),
4241 (121, "EREMOTEIO"),
4242 (122, "EDQUOT"),
4243 (123, "ENOMEDIUM"),
4244 (124, "EMEDIUMTYPE"),
4245 (125, "ECANCELED"),
4246 (126, "ENOKEY"),
4247 (127, "EKEYEXPIRED"),
4248 (128, "EKEYREVOKED"),
4249 (129, "EKEYREJECTED"),
4250 (130, "EOWNERDEAD"),
4251 (131, "ENOTRECOVERABLE"),
4252 (132, "ERFKILL"),
4253 (133, "EHWPOISON"),
4254];
4255
4256fn errno_name(raw: &str) -> Option<String> {
4257 let errno = raw.parse::<u32>().ok()?;
4258 let name = ERRNO_NAMES
4259 .iter()
4260 .find_map(|(candidate, name)| (*candidate == errno).then_some(*name))?;
4261 Some(format!("{errno} ({name})"))
4262}
4263
4264fn cap_effective_display(raw: &str) -> String {
4265 if !raw.bytes().next().is_some_and(|byte| byte.is_ascii_digit()) {
4266 return raw.to_string();
4267 }
4268 let Ok(value) = u64::from_str_radix(raw, 16) else {
4269 return raw.to_string();
4270 };
4271 if value == 0 {
4272 return raw.to_string();
4273 }
4274 const CAPABILITIES: &[&str] = &[
4275 "CHOWN",
4276 "DAC_OVERRIDE",
4277 "DAC_READ_SEARCH",
4278 "FOWNER",
4279 "FSETID",
4280 "KILL",
4281 "SETGID",
4282 "SETUID",
4283 "SETPCAP",
4284 "LINUX_IMMUTABLE",
4285 "NET_BIND_SERVICE",
4286 "NET_BROADCAST",
4287 "NET_ADMIN",
4288 "NET_RAW",
4289 "IPC_LOCK",
4290 "IPC_OWNER",
4291 "SYS_MODULE",
4292 "SYS_RAWIO",
4293 "SYS_CHROOT",
4294 "SYS_PTRACE",
4295 "SYS_PACCT",
4296 "SYS_ADMIN",
4297 "SYS_BOOT",
4298 "SYS_NICE",
4299 "SYS_RESOURCE",
4300 "SYS_TIME",
4301 "SYS_TTY_CONFIG",
4302 "MKNOD",
4303 "LEASE",
4304 "AUDIT_WRITE",
4305 "AUDIT_CONTROL",
4306 "SETFCAP",
4307 "MAC_OVERRIDE",
4308 "MAC_ADMIN",
4309 "SYSLOG",
4310 "WAKE_ALARM",
4311 "BLOCK_SUSPEND",
4312 "AUDIT_READ",
4313 "PERFMON",
4314 "BPF",
4315 "CHECKPOINT_RESTORE",
4316 ];
4317 let names: Vec<&str> = CAPABILITIES
4318 .iter()
4319 .enumerate()
4320 .filter_map(|(index, name)| ((value & (1u64 << index)) != 0).then_some(*name))
4321 .collect();
4322 if names.is_empty() {
4323 raw.to_string()
4324 } else {
4325 format!("{raw} ({})", names.join(" | "))
4326 }
4327}
4328
4329fn systemd_field_display_value(
4330 context: &DisplayContext,
4331 scope: DisplayScope,
4332 field: &str,
4333 value: &[u8],
4334 resolve_user_group_names: bool,
4335) -> Value {
4336 let raw = String::from_utf8_lossy(value);
4337 match field {
4338 "PRIORITY" => Value::String(priority_name(&raw).unwrap_or(&raw).to_string()),
4339 "SYSLOG_FACILITY" => Value::String(syslog_facility_name(&raw).unwrap_or(&raw).to_string()),
4340 "ERRNO" => Value::String(errno_name(&raw).unwrap_or_else(|| raw.to_string())),
4341 "MESSAGE_ID" => Value::String(match (message_id_name(&raw), scope) {
4342 (Some(name), DisplayScope::Data) => format!("{raw} ({name})"),
4343 (Some(name), DisplayScope::Facet | DisplayScope::Histogram) => name.to_string(),
4344 (None, _) => raw.into_owned(),
4345 }),
4346 "_BOOT_ID" => Value::String(match (context.boot_first_realtime.get(value), scope) {
4347 (Some(timestamp), DisplayScope::Data) => format!(
4348 "{} ({}) ",
4349 raw,
4350 format_realtime_usec(*timestamp, TimestampPrecision::Seconds)
4351 ),
4352 (Some(timestamp), DisplayScope::Facet | DisplayScope::Histogram) => {
4353 format_realtime_usec(*timestamp, TimestampPrecision::Seconds)
4354 }
4355 (None, _) => raw.into_owned(),
4356 }),
4357 "_UID"
4358 | "_SYSTEMD_OWNER_UID"
4359 | "OBJECT_SYSTEMD_OWNER_UID"
4360 | "OBJECT_UID"
4361 | "_AUDIT_LOGINUID"
4362 | "OBJECT_AUDIT_LOGINUID" => {
4363 if resolve_user_group_names {
4364 Value::String(cached_uid_display(context, raw.as_ref()))
4365 } else {
4366 Value::String(raw.into_owned())
4367 }
4368 }
4369 "_GID" | "OBJECT_GID" => {
4370 if resolve_user_group_names {
4371 Value::String(cached_gid_display(context, raw.as_ref()))
4372 } else {
4373 Value::String(raw.into_owned())
4374 }
4375 }
4376 "_CAP_EFFECTIVE" => Value::String(cap_effective_display(&raw)),
4377 "_SOURCE_REALTIME_TIMESTAMP" => Value::String(match raw.parse::<u64>() {
4378 Ok(timestamp) if timestamp != 0 => {
4379 format!(
4380 "{} ({})",
4381 raw,
4382 format_realtime_usec(timestamp, TimestampPrecision::Micros)
4383 )
4384 }
4385 _ => raw.into_owned(),
4386 }),
4387 _ => Value::String(raw.into_owned()),
4388 }
4389}
4390
4391fn cached_uid_display(context: &DisplayContext, raw: &str) -> String {
4392 if let Some(value) = context.uid_display_cache.borrow().get(raw) {
4393 return value.clone();
4394 }
4395 let value = resolve_uid_name(raw).unwrap_or_else(|| raw.to_string());
4396 context
4397 .uid_display_cache
4398 .borrow_mut()
4399 .insert(raw.to_string(), value.clone());
4400 value
4401}
4402
4403fn cached_gid_display(context: &DisplayContext, raw: &str) -> String {
4404 if let Some(value) = context.gid_display_cache.borrow().get(raw) {
4405 return value.clone();
4406 }
4407 let value = resolve_gid_name(raw).unwrap_or_else(|| raw.to_string());
4408 context
4409 .gid_display_cache
4410 .borrow_mut()
4411 .insert(raw.to_string(), value.clone());
4412 value
4413}
4414
4415#[cfg(unix)]
4416fn resolve_uid_name(raw: &str) -> Option<String> {
4417 let uid = raw.parse::<libc::uid_t>().ok()?;
4418 let mut pwd = std::mem::MaybeUninit::<libc::passwd>::uninit();
4419 let mut result = std::ptr::null_mut();
4420 let mut buffer = vec![0i8; 16_384];
4421 let rc = unsafe {
4422 libc::getpwuid_r(
4423 uid,
4424 pwd.as_mut_ptr(),
4425 buffer.as_mut_ptr(),
4426 buffer.len(),
4427 &mut result,
4428 )
4429 };
4430 if rc != 0 || result.is_null() {
4431 return None;
4432 }
4433 let pwd = unsafe { pwd.assume_init() };
4434 Some(
4435 unsafe { CStr::from_ptr(pwd.pw_name) }
4436 .to_string_lossy()
4437 .into_owned(),
4438 )
4439}
4440
4441#[cfg(not(unix))]
4442fn resolve_uid_name(_raw: &str) -> Option<String> {
4443 None
4444}
4445
4446#[cfg(unix)]
4447fn resolve_gid_name(raw: &str) -> Option<String> {
4448 let gid = raw.parse::<libc::gid_t>().ok()?;
4449 let mut grp = std::mem::MaybeUninit::<libc::group>::uninit();
4450 let mut result = std::ptr::null_mut();
4451 let mut buffer = vec![0i8; 16_384];
4452 let rc = unsafe {
4453 libc::getgrgid_r(
4454 gid,
4455 grp.as_mut_ptr(),
4456 buffer.as_mut_ptr(),
4457 buffer.len(),
4458 &mut result,
4459 )
4460 };
4461 if rc != 0 || result.is_null() {
4462 return None;
4463 }
4464 let grp = unsafe { grp.assume_init() };
4465 Some(
4466 unsafe { CStr::from_ptr(grp.gr_name) }
4467 .to_string_lossy()
4468 .into_owned(),
4469 )
4470}
4471
4472#[cfg(not(unix))]
4473fn resolve_gid_name(_raw: &str) -> Option<String> {
4474 None
4475}
4476
4477const MESSAGE_ID_NAMES: &[(&str, &str)] = &[
4478 ("f77379a8490b408bbe5f6940505a777b", "Journal started"),
4479 ("d93fb3c9c24d451a97cea615ce59c00b", "Journal stopped"),
4480 (
4481 "a596d6fe7bfa4994828e72309e95d61e",
4482 "Journal messages suppressed",
4483 ),
4484 (
4485 "e9bf28e6e834481bb6f48f548ad13606",
4486 "Journal messages missed",
4487 ),
4488 (
4489 "ec387f577b844b8fa948f33cad9a75e6",
4490 "Journal disk space usage",
4491 ),
4492 ("fc2e22bc6ee647b6b90729ab34a250b1", "Coredump"),
4493 ("5aadd8e954dc4b1a8c954d63fd9e1137", "Coredump truncated"),
4494 ("1f4e0a44a88649939aaea34fc6da8c95", "Backtrace"),
4495 ("8d45620c1a4348dbb17410da57c60c66", "User Session created"),
4496 (
4497 "3354939424b4456d9802ca8333ed424a",
4498 "User Session terminated",
4499 ),
4500 ("fcbefc5da23d428093f97c82a9290f7b", "Seat started"),
4501 ("e7852bfe46784ed0accde04bc864c2d5", "Seat removed"),
4502 (
4503 "24d8d4452573402496068381a6312df2",
4504 "VM or container started",
4505 ),
4506 (
4507 "58432bd3bace477cb514b56381b8a758",
4508 "VM or container stopped",
4509 ),
4510 ("c7a787079b354eaaa9e77b371893cd27", "Time change"),
4511 ("45f82f4aef7a4bbf942ce861d1f20990", "Timezone change"),
4512 (
4513 "50876a9db00f4c40bde1a2ad381c3a1b",
4514 "System configuration issues",
4515 ),
4516 (
4517 "b07a249cd024414a82dd00cd181378ff",
4518 "System start-up completed",
4519 ),
4520 (
4521 "eed00a68ffd84e31882105fd973abdd1",
4522 "User start-up completed",
4523 ),
4524 ("6bbd95ee977941e497c48be27c254128", "Sleep start"),
4525 ("8811e6df2a8e40f58a94cea26f8ebf14", "Sleep stop"),
4526 (
4527 "98268866d1d54a499c4e98921d93bc40",
4528 "System shutdown initiated",
4529 ),
4530 (
4531 "c14aaf76ec284a5fa1f105f88dfb061c",
4532 "System factory reset initiated",
4533 ),
4534 ("d9ec5e95e4b646aaaea2fd05214edbda", "Container init crashed"),
4535 (
4536 "3ed0163e868a4417ab8b9e210407a96c",
4537 "System reboot failed after crash",
4538 ),
4539 ("645c735537634ae0a32b15a7c6cba7d4", "Init execution froze"),
4540 (
4541 "5addb3a06a734d3396b794bf98fb2d01",
4542 "Init crashed no coredump",
4543 ),
4544 ("5c9e98de4ab94c6a9d04d0ad793bd903", "Init crashed no fork"),
4545 (
4546 "5e6f1f5e4db64a0eaee3368249d20b94",
4547 "Init crashed unknown signal",
4548 ),
4549 (
4550 "83f84b35ee264f74a3896a9717af34cb",
4551 "Init crashed systemd signal",
4552 ),
4553 (
4554 "3a73a98baf5b4b199929e3226c0be783",
4555 "Init crashed process signal",
4556 ),
4557 (
4558 "2ed18d4f78ca47f0a9bc25271c26adb4",
4559 "Init crashed waitpid failed",
4560 ),
4561 (
4562 "56b1cd96f24246c5b607666fda952356",
4563 "Init crashed coredump failed",
4564 ),
4565 ("4ac7566d4d7548f4981f629a28f0f829", "Init crashed coredump"),
4566 (
4567 "38e8b1e039ad469291b18b44c553a5b7",
4568 "Crash shell failed to fork",
4569 ),
4570 (
4571 "872729b47dbe473eb768ccecd477beda",
4572 "Crash shell failed to execute",
4573 ),
4574 ("658a67adc1c940b3b3316e7e8628834a", "Selinux failed"),
4575 ("e6f456bd92004d9580160b2207555186", "Battery low warning"),
4576 (
4577 "267437d33fdd41099ad76221cc24a335",
4578 "Battery low powering off",
4579 ),
4580 (
4581 "79e05b67bc4545d1922fe47107ee60c5",
4582 "Manager mainloop failed",
4583 ),
4584 ("dbb136b10ef4457ba47a795d62f108c9", "Manager no xdgdir path"),
4585 (
4586 "ed158c2df8884fa584eead2d902c1032",
4587 "Init failed to drop capability bounding set of usermode",
4588 ),
4589 (
4590 "42695b500df048298bee37159caa9f2e",
4591 "Init failed to drop capability bounding set",
4592 ),
4593 (
4594 "bfc2430724ab44499735b4f94cca9295",
4595 "User manager can't disable new privileges",
4596 ),
4597 (
4598 "59288af523be43a28d494e41e26e4510",
4599 "Manager failed to start default target",
4600 ),
4601 (
4602 "689b4fcc97b4486ea5da92db69c9e314",
4603 "Manager failed to isolate default target",
4604 ),
4605 (
4606 "5ed836f1766f4a8a9fc5da45aae23b29",
4607 "Manager failed to collect passed file descriptors",
4608 ),
4609 (
4610 "6a40fbfbd2ba4b8db02fb40c9cd090d7",
4611 "Init failed to fix up environment variables",
4612 ),
4613 (
4614 "0e54470984ac419689743d957a119e2e",
4615 "Manager failed to allocate",
4616 ),
4617 (
4618 "d67fa9f847aa4b048a2ae33535331adb",
4619 "Manager failed to write Smack",
4620 ),
4621 (
4622 "af55a6f75b544431b72649f36ff6d62c",
4623 "System shutdown critical error",
4624 ),
4625 (
4626 "d18e0339efb24a068d9c1060221048c2",
4627 "Init failed to fork off valgrind",
4628 ),
4629 ("7d4958e842da4a758f6c1cdc7b36dcc5", "Unit starting"),
4630 ("39f53479d3a045ac8e11786248231fbf", "Unit started"),
4631 ("be02cf6855d2428ba40df7e9d022f03d", "Unit failed"),
4632 ("de5b426a63be47a7b6ac3eaac82e2f6f", "Unit stopping"),
4633 ("9d1aaa27d60140bd96365438aad20286", "Unit stopped"),
4634 ("d34d037fff1847e6ae669a370e694725", "Unit reloading"),
4635 ("7b05ebc668384222baa8881179cfda54", "Unit reloaded"),
4636 ("5eb03494b6584870a536b337290809b3", "Unit restart scheduled"),
4637 ("ae8f7b866b0347b9af31fe1c80b127c0", "Unit resources"),
4638 ("7ad2d189f7e94e70a38c781354912448", "Unit success"),
4639 ("0e4284a0caca4bfc81c0bb6786972673", "Unit skipped"),
4640 ("d9b373ed55a64feb8242e02dbe79a49c", "Unit failure result"),
4641 (
4642 "641257651c1b4ec9a8624d7a40a9e1e7",
4643 "Process execution failed",
4644 ),
4645 ("98e322203f7a4ed290d09fe03c09fe15", "Unit process exited"),
4646 ("0027229ca0644181a76c4e92458afa2e", "Syslog forward missed"),
4647 (
4648 "1dee0369c7fc4736b7099b38ecb46ee7",
4649 "Mount point is not empty",
4650 ),
4651 ("d989611b15e44c9dbf31e3c81256e4ed", "Unit oomd kill"),
4652 ("fe6faa94e7774663a0da52717891d8ef", "Unit out of memory"),
4653 ("b72ea4a2881545a0b50e200e55b9b06f", "Lid opened"),
4654 ("b72ea4a2881545a0b50e200e55b9b070", "Lid closed"),
4655 ("f5f416b862074b28927a48c3ba7d51ff", "System docked"),
4656 ("51e171bd585248568110144c517cca53", "System undocked"),
4657 ("b72ea4a2881545a0b50e200e55b9b071", "Power key"),
4658 ("3e0117101eb243c1b9a50db3494ab10b", "Power key long press"),
4659 ("9fa9d2c012134ec385451ffe316f97d0", "Reboot key"),
4660 ("f1c59a58c9d943668965c337caec5975", "Reboot key long press"),
4661 ("b72ea4a2881545a0b50e200e55b9b072", "Suspend key"),
4662 ("bfdaf6d312ab4007bc1fe40a15df78e8", "Suspend key long press"),
4663 ("b72ea4a2881545a0b50e200e55b9b073", "Hibernate key"),
4664 (
4665 "167836df6f7f428e98147227b2dc8945",
4666 "Hibernate key long press",
4667 ),
4668 ("c772d24e9a884cbeb9ea12625c306c01", "Invalid configuration"),
4669 (
4670 "1675d7f172174098b1108bf8c7dc8f5d",
4671 "DNSSEC validation failed",
4672 ),
4673 (
4674 "4d4408cfd0d144859184d1e65d7c8a65",
4675 "DNSSEC trust anchor revoked",
4676 ),
4677 ("36db2dfa5a9045e1bd4af5f93e1cf057", "DNSSEC turned off"),
4678 ("b61fdac612e94b9182285b998843061f", "Username unsafe"),
4679 (
4680 "1b3bb94037f04bbf81028e135a12d293",
4681 "Mount point path not suitable",
4682 ),
4683 (
4684 "010190138f494e29a0ef6669749531aa",
4685 "Device path not suitable",
4686 ),
4687 ("b480325f9c394a7b802c231e51a2752c", "Nobody user unsuitable"),
4688 (
4689 "1c0454c1bd2241e0ac6fefb4bc631433",
4690 "Systemd udev settle deprecated",
4691 ),
4692 ("7c8a41f37b764941a0e1780b1be2f037", "Time initial sync"),
4693 ("7db73c8af0d94eeb822ae04323fe6ab6", "Time initial bump"),
4694 ("9e7066279dc8403da79ce4b1a69064b2", "Shutdown scheduled"),
4695 ("249f6fb9e6e2428c96f3f0875681ffa3", "Shutdown canceled"),
4696 ("3f7d5ef3e54f4302b4f0b143bb270cab", "TPM PCR Extended"),
4697 ("f9b0be465ad540d0850ad32172d57c21", "Memory Trimmed"),
4698 ("a8fa8dacdb1d443e9503b8be367a6adb", "SysV Service Found"),
4699 (
4700 "187c62eb1e7f463bb530394f52cb090f",
4701 "Portable Service attached",
4702 ),
4703 (
4704 "76c5c754d628490d8ecba4c9d042112b",
4705 "Portable Service detached",
4706 ),
4707 (
4708 "9cf56b8baf9546cf9478783a8de42113",
4709 "systemd-networkd sysctl changed by foreign process",
4710 ),
4711 (
4712 "ad7089f928ac4f7ea00c07457d47ba8a",
4713 "SRK into TPM authorization failure",
4714 ),
4715 (
4716 "b2bcbaf5edf948e093ce50bbea0e81ec",
4717 "Secure Attention Key (SAK) was pressed",
4718 ),
4719 ("7fc63312330b479bb32e598d47cef1a8", "dbus activate no unit"),
4720 (
4721 "ee9799dab1e24d81b7bee7759a543e1b",
4722 "dbus activate masked unit",
4723 ),
4724 ("a0fa58cafd6f4f0c8d003d16ccf9e797", "dbus broker exited"),
4725 ("c8c6cde1c488439aba371a664353d9d8", "dbus dirwatch"),
4726 ("8af3357071af4153af414daae07d38e7", "dbus dispatch stats"),
4727 ("199d4300277f495f84ba4028c984214c", "dbus no sopeergroup"),
4728 (
4729 "b209c0d9d1764ab38d13b8e00d1784d6",
4730 "dbus protocol violation",
4731 ),
4732 ("6fa70fa776044fa28be7a21daf42a108", "dbus receive failed"),
4733 (
4734 "0ce0fa61d1a9433dabd67417f6b8e535",
4735 "dbus service failed open",
4736 ),
4737 ("24dc708d9e6a4226a3efe2033bb744de", "dbus service invalid"),
4738 ("f15d2347662d483ea9bcd8aa1a691d28", "dbus sighup"),
4739 (
4740 "0ce153587afa4095832d233c17a88001",
4741 "Gnome SM startup succeeded",
4742 ),
4743 (
4744 "10dd2dc188b54a5e98970f56499d1f73",
4745 "Gnome SM unrecoverable failure",
4746 ),
4747 ("f3ea493c22934e26811cd62abe8e203a", "Gnome shell started"),
4748 ("c7b39b1e006b464599465e105b361485", "Flatpak cache"),
4749 ("75ba3deb0af041a9a46272ff85d9e73e", "Flathub pulls"),
4750 ("f02bce89a54e4efab3a94a797d26204a", "Flathub pull errors"),
4751 ("dd11929c788e48bdbb6276fb5f26b08a", "Boltd starting"),
4752 ("1e6061a9fbd44501b3ccc368119f2b69", "Netdata startup"),
4753 (
4754 "ed4cdb8f1beb4ad3b57cb3cae2d162fa",
4755 "Netdata connection from child",
4756 ),
4757 (
4758 "6e2e3839067648968b646045dbf28d66",
4759 "Netdata connection to parent",
4760 ),
4761 (
4762 "9ce0cb58ab8b44df82c4bf1ad9ee22de",
4763 "Netdata alert transition",
4764 ),
4765 (
4766 "6db0018e83e34320ae2a659d78019fb7",
4767 "Netdata alert notification",
4768 ),
4769 ("23e93dfccbf64e11aac858b9410d8a82", "Netdata fatal message"),
4770 (
4771 "8ddaf5ba33a74078b609250db1e951f3",
4772 "Sensor state transition",
4773 ),
4774 (
4775 "ec87a56120d5431bace51e2fb8bba243",
4776 "Netdata log flood protection",
4777 ),
4778 (
4779 "acb33cb95778476baac702eb7e4e151d",
4780 "Netdata Cloud connection",
4781 ),
4782 (
4783 "d1f59606dd4d41e3b217a0cfcae8e632",
4784 "Netdata extreme cardinality",
4785 ),
4786 ("02f47d350af5449197bf7a95b605a468", "Netdata exit reason"),
4787 (
4788 "4fdf40816c124623a032b7fe73beacb8",
4789 "Netdata dynamic configuration",
4790 ),
4791];
4792
4793fn message_id_name(raw: &str) -> Option<&'static str> {
4794 MESSAGE_ID_NAMES
4795 .iter()
4796 .find_map(|(candidate, name)| (*candidate == raw).then_some(*name))
4797}
4798
4799#[cfg(test)]
4800mod tests {
4801 use super::*;
4802 use crate::ExplorerHistogramBucket;
4803 use journal_core::file::{JournalFile, JournalFileOptions, JournalWriter, MmapMut};
4804 use journal_core::repository::File as RepoFile;
4805 use std::cell::Cell;
4806 use std::collections::HashMap;
4807 use tempfile::TempDir;
4808
4809 #[derive(Default)]
4810 struct TestNetdataState {
4811 metadata: HashMap<PathBuf, NetdataJournalFileMetadata>,
4812 updates: Vec<(PathBuf, u64)>,
4813 }
4814
4815 impl NetdataFunctionState for TestNetdataState {
4816 fn file_metadata(&self, path: &Path) -> Option<NetdataJournalFileMetadata> {
4817 self.metadata.get(path).cloned()
4818 }
4819
4820 fn update_file_journal_vs_realtime_delta_usec(&mut self, path: &Path, delta_usec: u64) {
4821 self.updates.push((path.to_path_buf(), delta_usec));
4822 }
4823 }
4824
4825 fn test_uuid(seed: u8) -> uuid::Uuid {
4826 uuid::Uuid::from_bytes([seed; 16])
4827 }
4828
4829 fn write_netdata_test_journal(directory: &std::path::Path, count: usize) {
4830 write_named_netdata_test_journal(
4831 directory,
4832 "netdata-api-test.journal",
4833 count,
4834 1_700_000_000_000_000,
4835 );
4836 }
4837
4838 fn write_named_netdata_test_journal(
4839 directory: &std::path::Path,
4840 name: &str,
4841 count: usize,
4842 start_realtime_usec: u64,
4843 ) {
4844 write_stepped_netdata_test_journal(directory, name, count, start_realtime_usec, 1);
4845 }
4846
4847 fn write_stepped_netdata_test_journal(
4848 directory: &std::path::Path,
4849 name: &str,
4850 count: usize,
4851 start_realtime_usec: u64,
4852 step_realtime_usec: u64,
4853 ) {
4854 std::fs::create_dir_all(directory).expect("create journal dir");
4855 let path = directory.join(name);
4856 let repo_file = RepoFile::from_path(&path).expect("repo file");
4857 let options = JournalFileOptions::new(test_uuid(1), test_uuid(2), test_uuid(3));
4858 let mut file = JournalFile::<MmapMut>::create(&repo_file, options).expect("create journal");
4859 let mut writer = JournalWriter::new(&mut file, 1, test_uuid(4)).expect("writer");
4860 for index in 0..count {
4861 let message = format!("MESSAGE=row-{index}");
4862 let service = if index % 2 == 0 {
4863 b"SERVICE=even".as_slice()
4864 } else {
4865 b"SERVICE=odd".as_slice()
4866 };
4867 let payloads: [&[u8]; 3] = [message.as_bytes(), service, b"PRIORITY=6"];
4868 let realtime = start_realtime_usec
4869 .saturating_add((index as u64).saturating_mul(step_realtime_usec));
4870 writer
4871 .add_entry(&mut file, &payloads, realtime, realtime)
4872 .expect("write entry");
4873 }
4874 file.sync().expect("sync journal");
4875 }
4876
4877 struct NetdataCollectedPages {
4878 messages: Vec<String>,
4879 timestamps: Vec<u64>,
4880 }
4881
4882 fn collect_netdata_pages(
4883 directory: &std::path::Path,
4884 direction: Direction,
4885 page_size: usize,
4886 ) -> NetdataCollectedPages {
4887 let mut out = NetdataCollectedPages {
4888 messages: Vec::new(),
4889 timestamps: Vec::new(),
4890 };
4891 let mut anchor = None;
4892 for page in 0..100 {
4893 let mut request = json!({
4894 "after": 1_700_000_000,
4895 "before": 1_800_000_000,
4896 "last": page_size,
4897 "direction": direction_name(direction),
4898 "data_only": true,
4899 });
4900 if let Some(anchor) = anchor {
4901 request["anchor"] = Value::from(anchor);
4902 }
4903 let response = run_netdata_contract_request(directory, request);
4904 assert_eq!(
4905 response["status"], 200,
4906 "page {page} response = {response:#}"
4907 );
4908 let messages = response_column_strings(&response, "MESSAGE");
4909 let timestamps = response_column_u64s(&response, "timestamp");
4910 assert_eq!(messages.len(), timestamps.len());
4911 if messages.is_empty() {
4912 break;
4913 }
4914 out.messages.extend(messages);
4915 out.timestamps.extend(timestamps.iter().copied());
4916 anchor = Some(match direction {
4917 Direction::Forward => timestamps[0],
4918 Direction::Backward => *timestamps.last().expect("page timestamp"),
4919 });
4920 if timestamps.len() < page_size {
4921 break;
4922 }
4923 }
4924 assert!(!out.messages.is_empty(), "pagination returned no rows");
4925 out
4926 }
4927
4928 fn run_netdata_contract_request(directory: &std::path::Path, request: Value) -> Value {
4929 NetdataJournalFunction::systemd_journal_plugin_compatible()
4930 .run_directory_request_json_with_options(
4931 directory,
4932 &request,
4933 NetdataFunctionRunOptions::from_timeout_seconds(0),
4934 )
4935 .expect("run function")
4936 }
4937
4938 fn direction_name(direction: Direction) -> &'static str {
4939 match direction {
4940 Direction::Forward => "forward",
4941 Direction::Backward => "backward",
4942 }
4943 }
4944
4945 fn response_column_strings(response: &Value, field: &str) -> Vec<String> {
4946 let index = response_column_index(response, field);
4947 response["data"]
4948 .as_array()
4949 .expect("data")
4950 .iter()
4951 .map(|row| {
4952 row.as_array().expect("row")[index]
4953 .as_str()
4954 .expect("string cell")
4955 .to_string()
4956 })
4957 .collect()
4958 }
4959
4960 fn response_column_u64s(response: &Value, field: &str) -> Vec<u64> {
4961 let index = response_column_index(response, field);
4962 response["data"]
4963 .as_array()
4964 .expect("data")
4965 .iter()
4966 .map(|row| {
4967 row.as_array().expect("row")[index]
4968 .as_u64()
4969 .expect("u64 cell")
4970 })
4971 .collect()
4972 }
4973
4974 fn response_column_index(response: &Value, field: &str) -> usize {
4975 response["columns"][field]["index"]
4976 .as_u64()
4977 .expect("column index") as usize
4978 }
4979
4980 fn response_facet_count(response: &Value, key: &str, field: &str, value: &str) -> u64 {
4981 for facet in response[key].as_array().expect("facets") {
4982 if facet["id"] != field {
4983 continue;
4984 }
4985 for option in facet["options"].as_array().expect("facet options") {
4986 if option["id"] == value {
4987 return option["count"].as_u64().expect("facet count");
4988 }
4989 }
4990 panic!("{key} facet {field} missing value {value}");
4991 }
4992 panic!("{key} missing facet {field}");
4993 }
4994
4995 fn response_histogram_total(response: &Value, key: &str, value: &str) -> u64 {
4996 let chart = &response[key]["chart"];
4997 let names = chart["view"]["dimensions"]["names"]
4998 .as_array()
4999 .expect("histogram names");
5000 let dimension_index = names
5001 .iter()
5002 .position(|name| name.as_str() == Some(value))
5003 .expect("histogram dimension");
5004 chart["result"]["data"]
5005 .as_array()
5006 .expect("histogram data")
5007 .iter()
5008 .map(|point| {
5009 point.as_array().expect("histogram point")[dimension_index + 1]
5010 .as_array()
5011 .expect("histogram cell")[0]
5012 .as_u64()
5013 .unwrap_or_default()
5014 })
5015 .sum()
5016 }
5017
5018 fn assert_unique_messages(messages: &[String]) {
5019 let mut seen = HashSet::new();
5020 for message in messages {
5021 assert!(
5022 seen.insert(message),
5023 "duplicate message {message} in {messages:?}"
5024 );
5025 }
5026 }
5027
5028 #[test]
5029 fn parses_netdata_selections_as_and_fields_or_values() {
5030 let request = json!({
5031 "after": 200_000_000,
5032 "before": 200_000_100,
5033 "direction": "forward",
5034 "last": 25,
5035 "facets": ["PRIORITY"],
5036 "selections": {
5037 "PRIORITY": ["warning", "error"],
5038 "_HOSTNAME": ["node-a"],
5039 "__logs_sources": ["all-local-system-logs"],
5040 }
5041 });
5042
5043 let parsed = NetdataRequest::parse(&request, &NetdataFunctionConfig::systemd_journal())
5044 .expect("parse request");
5045 assert_eq!(parsed.after_realtime_usec, Some(200_000_000_000_000));
5046 assert_eq!(parsed.before_realtime_usec, Some(200_000_100_999_999));
5047 assert_eq!(parsed.direction, Direction::Forward);
5048 assert_eq!(parsed.limit, 25);
5049 assert_eq!(parsed.filters.len(), 2);
5050 assert_eq!(parsed.filters[0].field, b"PRIORITY");
5051 assert_eq!(parsed.filters[0].values, vec![b"4".to_vec(), b"3".to_vec()]);
5052 assert_eq!(parsed.filters[1].field, b"_HOSTNAME");
5053 assert_eq!(parsed.filters[1].values, vec![b"node-a".to_vec()]);
5054 }
5055
5056 #[test]
5057 fn netdata_last_one_keeps_echo_and_uses_effective_minimum_two() {
5058 let request = json!({
5059 "after": 200_000_000,
5060 "before": 200_000_100,
5061 "last": 1
5062 });
5063
5064 let parsed = NetdataRequest::parse(&request, &NetdataFunctionConfig::systemd_journal())
5065 .expect("parse request");
5066 let query =
5067 parsed.to_explorer_query(1, None, NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC);
5068
5069 assert_eq!(parsed.echo.get("last").and_then(Value::as_u64), Some(1));
5070 assert_eq!(parsed.limit, 2);
5071 assert_eq!(query.limit, 2);
5072 }
5073
5074 #[test]
5075 fn netdata_facet_counts_use_native_sliced_filter_semantics() {
5076 let request = json!({
5077 "after": 200_000_000,
5078 "before": 200_000_100,
5079 "facets": ["PRIORITY"],
5080 "selections": {
5081 "PRIORITY": ["warning"]
5082 }
5083 });
5084
5085 let parsed = NetdataRequest::parse(&request, &NetdataFunctionConfig::systemd_journal())
5086 .expect("parse request");
5087 let query =
5088 parsed.to_explorer_query(1, None, NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC);
5089
5090 assert!(!query.exclude_facet_field_filters);
5091 }
5092
5093 #[test]
5094 fn netdata_multi_filter_facet_counts_exclude_same_field_filter() {
5095 let request = json!({
5096 "after": 200_000_000,
5097 "before": 200_000_100,
5098 "facets": ["PRIORITY", "_BOOT_ID"],
5099 "selections": {
5100 "PRIORITY": ["warning"],
5101 "_BOOT_ID": ["738043aea7b3417cbc3e9941ad26f769"]
5102 }
5103 });
5104
5105 let parsed = NetdataRequest::parse(&request, &NetdataFunctionConfig::systemd_journal())
5106 .expect("parse request");
5107 let query =
5108 parsed.to_explorer_query(1, None, NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC);
5109
5110 assert!(query.exclude_facet_field_filters);
5111 }
5112
5113 #[test]
5114 fn parses_netdata_fts_query_like_simple_pattern() {
5115 let (terms, positives, negatives) =
5116 parse_fts_query_patterns(r"error|warning|!debug|escaped\|pipe|\!literal| a*B");
5117
5118 assert_eq!(
5119 positives,
5120 vec![
5121 b"error".to_vec(),
5122 b"warning".to_vec(),
5123 b"escaped|pipe".to_vec(),
5124 b"!literal".to_vec(),
5125 b" a*B".to_vec(),
5126 ]
5127 );
5128 assert_eq!(negatives, vec![b"debug".to_vec()]);
5129 assert_eq!(terms.len(), 6);
5130 assert!(!terms[0].negative);
5131 assert!(terms[2].negative);
5132 assert_eq!(
5133 terms[5],
5134 ExplorerFtsPattern {
5135 parts: vec![b" a".to_vec(), b"B".to_vec()],
5136 negative: false,
5137 }
5138 );
5139
5140 let request = json!({
5141 "query": r"alpha|!debug|needle\|pipe",
5142 "facets": ["PRIORITY"],
5143 });
5144 let parsed = NetdataRequest::parse(&request, &NetdataFunctionConfig::systemd_journal())
5145 .expect("parse request");
5146 let query =
5147 parsed.to_explorer_query(1, None, NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC);
5148 assert_eq!(
5149 query.fts_patterns,
5150 vec![b"alpha".to_vec(), b"needle|pipe".to_vec()]
5151 );
5152 assert_eq!(query.fts_negative_patterns, vec![b"debug".to_vec()]);
5153 assert_eq!(query.fts_terms.len(), 3);
5154 }
5155
5156 #[test]
5157 fn netdata_requests_never_enable_debug_row_traversal_column_collection() {
5158 let request = json!({
5159 "facets": ["PRIORITY", "_HOSTNAME"],
5160 "histogram": "PRIORITY",
5161 "last": 25
5162 });
5163
5164 let parsed = NetdataRequest::parse(&request, &NetdataFunctionConfig::systemd_journal())
5165 .expect("parse request");
5166 let query =
5167 parsed.to_explorer_query(1, None, NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC);
5168
5169 assert!(!query.debug_collect_column_fields_by_row_traversal);
5170 }
5171
5172 #[test]
5173 fn netdata_function_reports_requested_facet_groups_even_when_absent() {
5174 let dir = TempDir::new().expect("tempdir");
5175 write_netdata_test_journal(dir.path(), 10);
5176 let request = json!({
5177 "after": 1_700_000_000,
5178 "before": 1_700_000_010,
5179 "facets": ["SERVICE", "MISSING_FIELD"],
5180 "histogram": "SERVICE",
5181 "last": 5,
5182 "slice": true
5183 });
5184 let function = NetdataJournalFunction::systemd_journal_plugin_compatible();
5185
5186 let response = function
5187 .run_directory_request_json_with_options(
5188 dir.path(),
5189 &request,
5190 NetdataFunctionRunOptions::from_timeout_seconds(0),
5191 )
5192 .expect("run function");
5193
5194 let columns = response["columns"].as_object().expect("columns");
5195 assert!(columns.contains_key("SERVICE"));
5196 assert!(columns.contains_key("MISSING_FIELD"));
5197 let facets = response["facets"].as_array().expect("facets");
5198 assert_eq!(
5199 facets
5200 .iter()
5201 .filter_map(|facet| facet["id"].as_str())
5202 .collect::<Vec<_>>(),
5203 vec!["SERVICE", "MISSING_FIELD"]
5204 );
5205 assert_eq!(
5206 facets[1]["options"].as_array().expect("missing options"),
5207 &Vec::<Value>::new()
5208 );
5209 let accepted = response["accepted_params"]
5210 .as_array()
5211 .expect("accepted params");
5212 assert!(accepted.iter().any(|value| value == "SERVICE"));
5213 assert!(accepted.iter().any(|value| value == "MISSING_FIELD"));
5214 let histograms = response["available_histograms"]
5215 .as_array()
5216 .expect("available histograms");
5217 assert!(histograms.iter().any(|value| value["id"] == "SERVICE"));
5218 assert!(
5219 histograms
5220 .iter()
5221 .any(|value| value["id"] == "MISSING_FIELD")
5222 );
5223 }
5224
5225 #[test]
5226 fn netdata_function_reports_zero_count_existing_facets_for_empty_results() {
5227 let dir = TempDir::new().expect("tempdir");
5228 write_netdata_test_journal(dir.path(), 10);
5229 let request = json!({
5230 "after": 1_700_000_000,
5231 "before": 1_700_000_010,
5232 "facets": ["PRIORITY"],
5233 "histogram": "PRIORITY",
5234 "selections": {
5235 "SERVICE": ["missing"]
5236 },
5237 "last": 5,
5238 "slice": true
5239 });
5240 let function = NetdataJournalFunction::systemd_journal_plugin_compatible();
5241
5242 let response = function
5243 .run_directory_request_json_with_options(
5244 dir.path(),
5245 &request,
5246 NetdataFunctionRunOptions::from_timeout_seconds(0),
5247 )
5248 .expect("run function");
5249
5250 let facets = response["facets"].as_array().expect("facets");
5251 assert_eq!(facets.len(), 1);
5252 assert_eq!(facets[0]["id"], "PRIORITY");
5253 let options = facets[0]["options"].as_array().expect("options");
5254 assert!(options.iter().any(|option| {
5255 option["id"] == "6" && option["name"] == "info" && option["count"] == 0
5256 }));
5257 assert_eq!(response["items"]["matched"], 0);
5258 }
5259
5260 #[test]
5261 fn netdata_function_api_reports_progress() {
5262 let dir = TempDir::new().expect("tempdir");
5263 write_netdata_test_journal(dir.path(), 9_000);
5264 let request = json!({
5265 "after": 1_700_000_000,
5266 "before": 1_700_000_010,
5267 "facets": ["SERVICE"],
5268 "histogram": "SERVICE",
5269 "last": 0
5270 });
5271 let function = NetdataJournalFunction::systemd_journal_plugin_compatible();
5272 let mut reports = 0u64;
5273 let mut progress = |progress: NetdataFunctionProgress| {
5274 reports = reports.saturating_add(1);
5275 assert_eq!(progress.current_file, 1);
5276 assert_eq!(progress.total_files, 1);
5277 assert!(progress.stats.rows_examined <= 9_000);
5278 };
5279 let mut options = NetdataFunctionRunOptions::from_timeout_seconds(0);
5280 options.progress_interval = Duration::ZERO;
5281 options.progress_callback = Some(&mut progress);
5282
5283 let response = function
5284 .run_directory_request_json_with_options(dir.path(), &request, options)
5285 .expect("run function");
5286
5287 assert_eq!(response["status"], 200);
5288 assert!(reports > 0);
5289 assert_eq!(response["last_modified"], 1_700_000_000_008_999u64);
5290 }
5291
5292 #[test]
5293 fn netdata_function_api_reports_file_end_progress_for_small_scans() {
5294 let dir = TempDir::new().expect("tempdir");
5295 write_netdata_test_journal(dir.path(), 10);
5296 let request = json!({
5297 "after": 1_700_000_000,
5298 "before": 1_700_000_010,
5299 "facets": ["SERVICE"],
5300 "histogram": "SERVICE",
5301 "last": 0
5302 });
5303 let function = NetdataJournalFunction::systemd_journal_plugin_compatible();
5304 let mut reports = 0u64;
5305 let mut last_rows_examined = 0u64;
5306 let mut progress = |progress: NetdataFunctionProgress| {
5307 reports = reports.saturating_add(1);
5308 last_rows_examined = progress.stats.rows_examined;
5309 };
5310 let mut options = NetdataFunctionRunOptions::from_timeout_seconds(0);
5311 options.progress_callback = Some(&mut progress);
5312
5313 let response = function
5314 .run_directory_request_json_with_options(dir.path(), &request, options)
5315 .expect("run function");
5316
5317 assert_eq!(response["status"], 200);
5318 assert_eq!(reports, 1);
5319 assert_eq!(last_rows_examined, 10);
5320 }
5321
5322 #[test]
5323 fn netdata_function_progress_counts_only_query_files() {
5324 let dir = TempDir::new().expect("tempdir");
5325 write_named_netdata_test_journal(
5326 dir.path(),
5327 "old-window.journal",
5328 10,
5329 1_600_000_000_000_000,
5330 );
5331 write_named_netdata_test_journal(
5332 dir.path(),
5333 "current-window.journal",
5334 10,
5335 1_700_000_000_000_000,
5336 );
5337 let request = json!({
5338 "after": 1_700_000_000,
5339 "before": 1_700_000_010,
5340 "facets": ["SERVICE"],
5341 "histogram": "SERVICE",
5342 "last": 0
5343 });
5344 let function = NetdataJournalFunction::systemd_journal_plugin_compatible();
5345 let mut reports = Vec::new();
5346 let mut progress = |progress: NetdataFunctionProgress| {
5347 reports.push((progress.current_file, progress.total_files));
5348 };
5349 let mut options = NetdataFunctionRunOptions::from_timeout_seconds(0);
5350 options.progress_callback = Some(&mut progress);
5351
5352 let response = function
5353 .run_directory_request_json_with_options(dir.path(), &request, options)
5354 .expect("run function");
5355
5356 assert_eq!(response["status"], 200);
5357 assert_eq!(response["_journal_files"]["matched"], 1);
5358 assert_eq!(reports, vec![(1, 1)]);
5359 }
5360
5361 #[test]
5362 fn netdata_function_api_reports_cancellation() {
5363 let dir = TempDir::new().expect("tempdir");
5364 write_netdata_test_journal(dir.path(), 9_000);
5365 let request = json!({
5366 "after": 1_700_000_000,
5367 "before": 1_700_000_010,
5368 "facets": ["SERVICE"],
5369 "histogram": "SERVICE",
5370 "last": 0
5371 });
5372 let function = NetdataJournalFunction::systemd_journal_plugin_compatible();
5373 let is_cancelled = || true;
5374 let mut options = NetdataFunctionRunOptions::from_timeout_seconds(0);
5375 options.cancellation_callback = Some(&is_cancelled);
5376
5377 let response = function
5378 .run_directory_request_json_with_options(dir.path(), &request, options)
5379 .expect("run function");
5380
5381 assert_eq!(response["status"], 499);
5382 assert_eq!(response["errorMessage"], "Request cancelled.");
5383 assert_eq!(
5384 response.as_object().expect("response object").len(),
5385 2,
5386 "plugin-compatible function errors only include status and errorMessage"
5387 );
5388 }
5389
5390 #[test]
5391 fn netdata_function_api_cancels_during_active_scan() {
5392 let dir = TempDir::new().expect("tempdir");
5393 write_netdata_test_journal(dir.path(), 9_000);
5394 let request = json!({
5395 "after": 1_700_000_000,
5396 "before": 1_700_000_010,
5397 "facets": ["SERVICE"],
5398 "histogram": "SERVICE",
5399 "last": 0
5400 });
5401 let function = NetdataJournalFunction::systemd_journal_plugin_compatible();
5402 let should_cancel = Cell::new(false);
5403 let mut reports = 0u64;
5404 let mut progress = |progress: NetdataFunctionProgress| {
5405 reports = reports.saturating_add(1);
5406 if progress.stats.rows_examined > 0 {
5407 should_cancel.set(true);
5408 }
5409 };
5410 let is_cancelled = || should_cancel.get();
5411 let mut options = NetdataFunctionRunOptions::from_timeout_seconds(0);
5412 options.progress_interval = Duration::ZERO;
5413 options.progress_callback = Some(&mut progress);
5414 options.cancellation_callback = Some(&is_cancelled);
5415
5416 let response = function
5417 .run_directory_request_json_with_options(dir.path(), &request, options)
5418 .expect("run function");
5419
5420 assert_eq!(response["status"], 499);
5421 assert_eq!(response["errorMessage"], "Request cancelled.");
5422 assert!(reports > 0);
5423 assert!(should_cancel.get());
5424 }
5425
5426 #[test]
5427 fn netdata_function_api_honors_cancellation_after_final_file_progress() {
5428 let dir = TempDir::new().expect("tempdir");
5429 write_netdata_test_journal(dir.path(), 10);
5430 let request = json!({
5431 "after": 1_700_000_000,
5432 "before": 1_700_000_010,
5433 "facets": ["SERVICE"],
5434 "histogram": "SERVICE",
5435 "last": 0
5436 });
5437 let function = NetdataJournalFunction::systemd_journal_plugin_compatible();
5438 let should_cancel = Cell::new(false);
5439 let mut progress = |_progress: NetdataFunctionProgress| {
5440 should_cancel.set(true);
5441 };
5442 let is_cancelled = || should_cancel.get();
5443 let mut options = NetdataFunctionRunOptions::from_timeout_seconds(0);
5444 options.progress_callback = Some(&mut progress);
5445 options.cancellation_callback = Some(&is_cancelled);
5446
5447 let response = function
5448 .run_directory_request_json_with_options(dir.path(), &request, options)
5449 .expect("run function");
5450
5451 assert_eq!(response["status"], 499);
5452 assert_eq!(response["errorMessage"], "Request cancelled.");
5453 assert!(should_cancel.get());
5454 }
5455
5456 #[test]
5457 fn netdata_function_api_reports_timeout_as_partial_table() {
5458 let dir = TempDir::new().expect("tempdir");
5459 write_netdata_test_journal(dir.path(), 10);
5460 let request = json!({
5461 "after": 1_700_000_000,
5462 "before": 1_700_000_010,
5463 "facets": ["SERVICE"],
5464 "histogram": "SERVICE",
5465 "last": 0
5466 });
5467 let function = NetdataJournalFunction::systemd_journal_plugin_compatible();
5468 let options = NetdataFunctionRunOptions {
5469 timeout: Some(Duration::ZERO),
5470 ..NetdataFunctionRunOptions::default()
5471 };
5472
5473 let response = function
5474 .run_directory_request_json_with_options(dir.path(), &request, options)
5475 .expect("run function");
5476
5477 assert_eq!(response["status"], 200);
5478 assert_eq!(response["partial"], true);
5479 assert_eq!(response["message"]["status"], "warning");
5480 assert_eq!(
5481 response["message"]["title"],
5482 "Query timed-out, incomplete data. "
5483 );
5484 }
5485
5486 #[test]
5487 fn netdata_function_api_reports_sampling_counters() {
5488 let dir = TempDir::new().expect("tempdir");
5489 write_stepped_netdata_test_journal(
5490 dir.path(),
5491 "netdata-api-test.journal",
5492 5_000,
5493 1_700_000_000_000_000,
5494 1_000,
5495 );
5496 let request = json!({
5497 "after": 1_700_000_000,
5498 "before": 1_700_000_005,
5499 "facets": ["SERVICE"],
5500 "histogram": "SERVICE",
5501 "last": 5,
5502 "sampling": 20
5503 });
5504 let function = NetdataJournalFunction::systemd_journal_plugin_compatible();
5505
5506 let response = function
5507 .run_directory_request_json_with_options(
5508 dir.path(),
5509 &request,
5510 NetdataFunctionRunOptions::from_timeout_seconds(0),
5511 )
5512 .expect("run function");
5513
5514 assert_eq!(response["status"], 200);
5515 assert!(
5516 response["_sampling"]["sampled"]
5517 .as_u64()
5518 .unwrap_or_default()
5519 > 0
5520 );
5521 assert!(
5522 response["_sampling"]["unsampled"]
5523 .as_u64()
5524 .unwrap_or_default()
5525 > 0
5526 );
5527 assert!(
5528 response["_sampling"]["estimated"]
5529 .as_u64()
5530 .unwrap_or_default()
5531 > 0
5532 );
5533 assert_eq!(
5534 response["items"]["estimated"],
5535 response["_sampling"]["estimated"]
5536 );
5537 assert!(
5538 response["items"]["unsampled"].as_u64().unwrap_or_default()
5539 < response["_sampling"]["unsampled"]
5540 .as_u64()
5541 .unwrap_or_default()
5542 );
5543 assert_eq!(response["message"]["status"], "notice");
5544 }
5545
5546 #[test]
5547 fn netdata_function_api_disables_sampling_for_data_only() {
5548 let dir = TempDir::new().expect("tempdir");
5549 write_netdata_test_journal(dir.path(), 5_000);
5550 let request = json!({
5551 "after": 1_700_000_000,
5552 "before": 1_700_000_010,
5553 "data_only": true,
5554 "last": 5,
5555 "sampling": 20
5556 });
5557 let function = NetdataJournalFunction::systemd_journal_plugin_compatible();
5558
5559 let response = function
5560 .run_directory_request_json_with_options(
5561 dir.path(),
5562 &request,
5563 NetdataFunctionRunOptions::from_timeout_seconds(0),
5564 )
5565 .expect("run function");
5566
5567 assert_eq!(response["status"], 200);
5568 assert!(response.get("_sampling").is_none());
5569 }
5570
5571 #[test]
5572 fn normalizes_missing_time_window_to_last_hour_like_plugin() {
5573 assert_eq!(
5574 normalize_time_window(1_000_000_000, None, None),
5575 (Some(999_996_400_000_000), Some(1_000_000_000_999_999))
5576 );
5577 }
5578
5579 #[test]
5580 fn normalizes_inverted_time_window_like_plugin() {
5581 assert_eq!(
5582 normalize_time_window(1_000_000_000, Some(200_000_100), Some(200_000_000)),
5583 (Some(200_000_000_000_000), Some(200_000_100_999_999))
5584 );
5585 }
5586
5587 #[test]
5588 fn normalizes_equal_time_window_like_plugin() {
5589 assert_eq!(
5590 normalize_time_window(1_000_000_000, Some(200_000_000), Some(200_000_000)),
5591 (Some(199_996_400_000_000), Some(200_000_000_999_999))
5592 );
5593 }
5594
5595 #[test]
5596 fn normalizes_relative_time_window_like_plugin() {
5597 assert_eq!(
5598 normalize_time_window(1_000_000_000, Some(100), Some(200)),
5599 (Some(999_999_701_000_000), Some(999_999_800_999_999))
5600 );
5601 }
5602
5603 #[test]
5604 fn normalizes_missing_after_with_supplied_before_like_plugin() {
5605 assert_eq!(
5606 normalize_time_window(1_000_000_000, None, Some(200_000_000)),
5607 (Some(199_999_401_000_000), Some(200_000_000_999_999))
5608 );
5609 }
5610
5611 #[test]
5612 fn systemd_profile_transforms_priority_and_facility_for_display() {
5613 let profile = SystemdJournalProfile;
5614 let context = DisplayContext::default();
5615 assert_eq!(
5616 profile.field_display_value(&context, DisplayScope::Data, "PRIORITY", b"7"),
5617 json!("debug")
5618 );
5619 assert_eq!(
5620 profile.field_display_value(&context, DisplayScope::Data, "SYSLOG_FACILITY", b"3"),
5621 json!("daemon")
5622 );
5623 assert_eq!(priority_to_row_severity(b"3"), "critical");
5624 assert_eq!(priority_to_row_severity(b"6"), "normal");
5625 }
5626
5627 #[test]
5628 fn dynamic_process_name_matches_plugin_fallback_order() {
5629 let mut fields = BTreeMap::new();
5630 fields.insert("SYSLOG_IDENTIFIER".to_string(), vec![b"syslog".to_vec()]);
5631 fields.insert("_COMM".to_string(), vec![b"comm".to_vec()]);
5632 fields.insert("_PID".to_string(), vec![b"42".to_vec()]);
5633 fields.insert("SYSLOG_PID".to_string(), vec![b"99".to_vec()]);
5634 assert_eq!(dynamic_process_name(&fields), "syslog[42]");
5635
5636 fields.insert("CONTAINER_NAME".to_string(), vec![b"container".to_vec()]);
5637 assert_eq!(dynamic_process_name(&fields), "container[42]");
5638
5639 fields.remove("CONTAINER_NAME");
5640 fields.remove("SYSLOG_IDENTIFIER");
5641 fields.remove("_PID");
5642 assert_eq!(dynamic_process_name(&fields), "comm[-]");
5643
5644 fields.insert("_PID".to_string(), vec![Vec::new()]);
5645 assert_eq!(dynamic_process_name(&fields), "comm");
5646
5647 fields.remove("_COMM");
5648 fields.remove("_PID");
5649 fields.insert("_EXE".to_string(), vec![b"/usr/bin/app".to_vec()]);
5650 assert_eq!(dynamic_process_name(&fields), "-");
5651 }
5652
5653 #[test]
5654 fn facet_values_are_truncated_and_collapsed_like_plugin() {
5655 let prefix = vec![b'a'; NETDATA_FACET_MAX_VALUE_LENGTH];
5656 let mut first = prefix.clone();
5657 first.extend_from_slice(b"-first");
5658 let mut second = prefix.clone();
5659 second.extend_from_slice(b"-second");
5660
5661 let mut values = BTreeMap::new();
5662 add_netdata_facet_count(&mut values, &first, 2);
5663 add_netdata_facet_count(&mut values, &second, 3);
5664
5665 assert_eq!(values.len(), 1);
5666 assert_eq!(values.get(&prefix), Some(&5));
5667 }
5668
5669 #[test]
5670 fn histogram_values_are_truncated_and_collapsed_like_plugin() {
5671 let prefix = vec![b'b'; NETDATA_FACET_MAX_VALUE_LENGTH];
5672 let mut first = prefix.clone();
5673 first.extend_from_slice(b"-first");
5674 let mut second = prefix.clone();
5675 second.extend_from_slice(b"-second");
5676
5677 let mut values = HashMap::new();
5678 values.insert(first, 2);
5679 values.insert(second, 3);
5680 let histogram = ExplorerHistogram {
5681 field: b"TEST_FIELD".to_vec(),
5682 buckets: vec![ExplorerHistogramBucket {
5683 start_realtime_usec: 1_000_000,
5684 end_realtime_usec: 2_000_000,
5685 values,
5686 }],
5687 };
5688
5689 let function = NetdataJournalFunction::systemd_journal();
5690 let rendered = function.build_histogram(&DisplayContext::default(), &histogram, None);
5691 let labels = rendered["chart"]["result"]["labels"]
5692 .as_array()
5693 .expect("labels");
5694 assert_eq!(labels.len(), 2);
5695 assert_eq!(labels[1], Value::String(String::from_utf8(prefix).unwrap()));
5696 assert_eq!(rendered["chart"]["result"]["data"][0][1][0], json!(5));
5697 }
5698
5699 #[test]
5700 fn histogram_chart_metadata_includes_empty_dimension_arrays() {
5701 let function = NetdataJournalFunction::systemd_journal_plugin_compatible();
5702 let empty = function.build_histogram(
5703 &DisplayContext::default(),
5704 &ExplorerHistogram {
5705 field: b"TRAP_SEVERITY".to_vec(),
5706 buckets: vec![ExplorerHistogramBucket {
5707 start_realtime_usec: 1_700_000_000_000_000,
5708 end_realtime_usec: 1_700_000_005_000_000,
5709 values: HashMap::new(),
5710 }],
5711 },
5712 None,
5713 );
5714
5715 assert_eq!(empty["chart"]["view"]["dimensions"]["names"], json!([]));
5716 assert_eq!(empty["chart"]["view"]["dimensions"]["ids"], json!([]));
5717 assert_eq!(empty["chart"]["db"]["dimensions"]["names"], json!([]));
5718
5719 let mut values = HashMap::new();
5720 values.insert(b"warning".to_vec(), 7);
5721 let with_value = function.build_histogram(
5722 &DisplayContext::default(),
5723 &ExplorerHistogram {
5724 field: b"TRAP_SEVERITY".to_vec(),
5725 buckets: vec![ExplorerHistogramBucket {
5726 start_realtime_usec: 1_700_000_000_000_000,
5727 end_realtime_usec: 1_700_000_005_000_000,
5728 values,
5729 }],
5730 },
5731 None,
5732 );
5733
5734 assert_eq!(
5735 with_value["chart"]["view"]["dimensions"]["names"],
5736 json!(["warning"])
5737 );
5738 assert_eq!(
5739 with_value["chart"]["view"]["dimensions"]["sts"]["min"],
5740 json!([7])
5741 );
5742 }
5743
5744 #[test]
5745 fn duplicate_row_timestamps_match_plugin_direction_adjustment() {
5746 let mut backward = vec![
5747 test_located_row(100),
5748 test_located_row(100),
5749 test_located_row(100),
5750 test_located_row(90),
5751 ];
5752 make_row_timestamps_unique_same_file(&mut backward, Direction::Backward);
5753 assert_eq!(
5754 backward
5755 .iter()
5756 .map(|row| row.row.realtime_usec)
5757 .collect::<Vec<_>>(),
5758 vec![100, 99, 98, 90]
5759 );
5760
5761 let mut forward = vec![
5762 test_located_row(90),
5763 test_located_row(100),
5764 test_located_row(100),
5765 test_located_row(100),
5766 ];
5767 make_row_timestamps_unique_same_file(&mut forward, Direction::Forward);
5768 assert_eq!(
5769 forward
5770 .iter()
5771 .map(|row| row.row.realtime_usec)
5772 .collect::<Vec<_>>(),
5773 vec![90, 100, 101, 102]
5774 );
5775 }
5776
5777 #[test]
5778 fn sort_and_limit_retains_full_raw_boundary_group_when_last_cuts_through_it() {
5779 let mut rows = vec![
5780 test_located_row_in_file("c.journal", 100, "c0"),
5781 test_located_row_in_file("b.journal", 100, "b0"),
5782 test_located_row_in_file("a.journal", 100, "a0"),
5783 ];
5784 rows.sort_by(|left, right| {
5785 left.file_path
5786 .cmp(&right.file_path)
5787 .then_with(|| left.row.cursor.cmp(&right.row.cursor))
5788 });
5789 sort_rows_by_raw_realtime(&mut rows, Direction::Backward);
5790 retain_boundary_group(&mut rows, Direction::Backward, 2);
5791 make_row_timestamps_unique_same_file(&mut rows, Direction::Backward);
5792
5793 assert_eq!(rows.len(), 3);
5794 let messages: Vec<&str> = rows
5795 .iter()
5796 .map(|row| row.file_path.to_str().unwrap())
5797 .collect();
5798 assert_eq!(messages, vec!["a.journal", "b.journal", "c.journal"]);
5799 }
5800
5801 #[test]
5802 fn sort_and_limit_retains_full_raw_boundary_group_for_forward_query() {
5803 let mut rows = vec![
5804 test_located_row_in_file("c.journal", 100, "c0"),
5805 test_located_row_in_file("b.journal", 100, "b0"),
5806 test_located_row_in_file("a.journal", 100, "a0"),
5807 ];
5808 rows.sort_by(|left, right| {
5809 left.file_path
5810 .cmp(&right.file_path)
5811 .then_with(|| left.row.cursor.cmp(&right.row.cursor))
5812 });
5813 sort_rows_by_raw_realtime(&mut rows, Direction::Forward);
5814 retain_boundary_group(&mut rows, Direction::Forward, 2);
5815 make_row_timestamps_unique_same_file(&mut rows, Direction::Forward);
5816
5817 assert_eq!(rows.len(), 3);
5818 }
5819
5820 #[test]
5821 fn sort_and_limit_keeps_cross_file_equal_timestamps_for_anchor_continuity() {
5822 let mut rows = vec![
5823 test_located_row_in_file("c.journal", 100, "c0"),
5824 test_located_row_in_file("b.journal", 100, "b0"),
5825 test_located_row_in_file("a.journal", 100, "a0"),
5826 test_located_row_in_file("d.journal", 90, "d0"),
5827 ];
5828 sort_rows_by_raw_realtime(&mut rows, Direction::Backward);
5829 make_row_timestamps_unique_same_file(&mut rows, Direction::Backward);
5830
5831 let boundary_group: Vec<u64> = rows
5832 .iter()
5833 .filter(|row| row.row.realtime_usec == 100)
5834 .map(|row| row.row.realtime_usec)
5835 .collect();
5836 assert_eq!(boundary_group.len(), 3);
5837 }
5838
5839 #[test]
5840 fn sort_and_limit_keeps_same_file_duplicate_display_adjustment() {
5841 let mut rows = vec![
5842 test_located_row_in_file("a.journal", 100, "a0"),
5843 test_located_row_in_file("a.journal", 100, "a1"),
5844 test_located_row_in_file("a.journal", 100, "a2"),
5845 test_located_row_in_file("a.journal", 90, "a3"),
5846 ];
5847 sort_rows_by_raw_realtime(&mut rows, Direction::Backward);
5848 make_row_timestamps_unique_same_file(&mut rows, Direction::Backward);
5849
5850 let same_file_timestamps: Vec<u64> = rows
5851 .iter()
5852 .filter(|row| row.file_path == Path::new("a.journal"))
5853 .map(|row| row.row.realtime_usec)
5854 .collect();
5855 assert_eq!(same_file_timestamps, vec![100, 99, 98, 90]);
5856 }
5857
5858 #[test]
5859 fn backward_realtime_anchor_is_exclusive_for_non_data_only_requests() {
5860 let config = NetdataFunctionConfig::systemd_journal();
5861 let parsed = NetdataRequest::parse(
5862 &json!({
5863 "after": 1_700_000_000,
5864 "before": 1_700_000_010,
5865 "anchor": 1_700_000_005_000_000u64,
5866 "direction": "backward",
5867 "last": 5
5868 }),
5869 &config,
5870 )
5871 .expect("parse non-data-only backward request");
5872 let query =
5873 parsed.to_explorer_query(1, None, NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC);
5874
5875 assert_eq!(query.before_realtime_usec, Some(1_700_000_004_999_999));
5876 assert!(matches!(query.anchor, ExplorerAnchor::Auto));
5877 }
5878
5879 #[test]
5880 fn backward_realtime_anchor_remains_exclusive_for_data_only_requests() {
5881 let config = NetdataFunctionConfig::systemd_journal();
5882 let parsed = NetdataRequest::parse(
5883 &json!({
5884 "after": 1_700_000_000,
5885 "before": 1_700_000_010,
5886 "anchor": 1_700_000_005_000_000u64,
5887 "data_only": true,
5888 "direction": "backward",
5889 "last": 5
5890 }),
5891 &config,
5892 )
5893 .expect("parse data-only backward request");
5894 let query =
5895 parsed.to_explorer_query(1, None, NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC);
5896
5897 assert_eq!(query.before_realtime_usec, Some(1_700_000_004_999_999));
5898 assert!(matches!(query.anchor, ExplorerAnchor::Auto));
5899 }
5900
5901 #[test]
5902 fn sort_and_limit_clears_rows_when_limit_is_zero() {
5903 let mut rows = vec![
5904 test_located_row_in_file("a.journal", 100, "a0"),
5905 test_located_row_in_file("b.journal", 90, "b0"),
5906 ];
5907 sort_rows_by_raw_realtime(&mut rows, Direction::Backward);
5908 retain_boundary_group(&mut rows, Direction::Backward, 0);
5909
5910 assert!(rows.is_empty());
5911 }
5912
5913 #[test]
5914 fn page_window_counts_forward_anchor_like_netdata_facets() {
5915 let config = NetdataFunctionConfig::systemd_journal();
5916 let request = NetdataRequest::parse(
5917 &json!({
5918 "after": 1_700_000_000,
5919 "before": 1_700_000_010,
5920 "anchor": 1_700_000_005_000_000u64,
5921 "direction": "forward",
5922 "last": 2
5923 }),
5924 &config,
5925 )
5926 .expect("parse request");
5927 let mut window = NetdataPageWindow::for_request(&request);
5928
5929 for realtime_usec in [
5930 1_700_000_003_000_000,
5931 1_700_000_004_000_000,
5932 1_700_000_006_000_000,
5933 1_700_000_007_000_000,
5934 1_700_000_008_000_000,
5935 ] {
5936 window.observe(realtime_usec);
5937 }
5938
5939 let counters = window.counters();
5940 assert_eq!(counters.matched, 3);
5941 assert_eq!(counters.before, 1);
5942 assert_eq!(counters.after, 2);
5943 }
5944
5945 #[test]
5946 fn page_window_counts_backward_anchor_like_netdata_facets() {
5947 let config = NetdataFunctionConfig::systemd_journal();
5948 let request = NetdataRequest::parse(
5949 &json!({
5950 "after": 1_700_000_000,
5951 "before": 1_700_000_010,
5952 "anchor": 1_700_000_008_000_000u64,
5953 "direction": "backward",
5954 "last": 2
5955 }),
5956 &config,
5957 )
5958 .expect("parse request");
5959 let mut window = NetdataPageWindow::for_request(&request);
5960
5961 for realtime_usec in [
5962 1_700_000_009_000_000,
5963 1_700_000_007_000_000,
5964 1_700_000_006_000_000,
5965 1_700_000_005_000_000,
5966 ] {
5967 window.observe(realtime_usec);
5968 }
5969
5970 let counters = window.counters();
5971 assert_eq!(counters.matched, 3);
5972 assert_eq!(counters.before, 1);
5973 assert_eq!(counters.after, 1);
5974 }
5975
5976 #[test]
5977 fn page_window_counts_tail_anchor_like_netdata_facets() {
5978 let config = NetdataFunctionConfig::systemd_journal();
5979 let request = NetdataRequest::parse(
5980 &json!({
5981 "after": 1_700_000_000,
5982 "before": 1_700_000_010,
5983 "anchor": 1_700_000_008_000_000u64,
5984 "if_modified_since": 1_700_000_008_000_000u64,
5985 "data_only": true,
5986 "tail": true,
5987 "last": 2
5988 }),
5989 &config,
5990 )
5991 .expect("parse request");
5992 let mut window = NetdataPageWindow::for_request(&request);
5993
5994 for realtime_usec in [
5995 1_700_000_009_000_000,
5996 1_700_000_008_000_000,
5997 1_700_000_007_000_000,
5998 ] {
5999 window.observe(realtime_usec);
6000 }
6001
6002 let counters = window.counters();
6003 assert_eq!(counters.matched, 1);
6004 assert_eq!(counters.before, 0);
6005 assert_eq!(counters.after, 2);
6006 }
6007
6008 #[test]
6009 fn netdata_function_tail_anchor_with_newer_filtered_out_rows_returns_empty_200() {
6010 let dir = TempDir::new().expect("tempdir");
6011 let start_realtime_usec = 1_700_000_000_000_000;
6012 write_stepped_netdata_test_journal(
6013 dir.path(),
6014 "netdata-api-test.journal",
6015 2,
6016 start_realtime_usec,
6017 1_000_000,
6018 );
6019 let request = json!({
6020 "after": 1_700_000_000,
6021 "before": 1_700_000_002,
6022 "anchor": start_realtime_usec,
6023 "if_modified_since": start_realtime_usec,
6024 "data_only": true,
6025 "tail": true,
6026 "last": 5,
6027 "selections": {
6028 "SERVICE": ["even"]
6029 }
6030 });
6031 let function = NetdataJournalFunction::systemd_journal_plugin_compatible();
6032
6033 let response = function
6034 .run_directory_request_json_with_options(
6035 dir.path(),
6036 &request,
6037 NetdataFunctionRunOptions::from_timeout_seconds(0),
6038 )
6039 .expect("run function");
6040
6041 assert_eq!(response["status"], 200);
6042 assert_eq!(
6043 response_column_strings(&response, "MESSAGE"),
6044 Vec::<String>::new()
6045 );
6046 }
6047
6048 #[test]
6049 fn netdata_function_pages_with_anchor_without_duplicate_or_missing_rows() {
6050 let dir = TempDir::new().expect("tempdir");
6051 let start_realtime_usec = 1_700_000_000_000_000;
6052 write_stepped_netdata_test_journal(
6053 dir.path(),
6054 "paging-contract.journal",
6055 7,
6056 start_realtime_usec,
6057 1,
6058 );
6059
6060 let backward = collect_netdata_pages(dir.path(), Direction::Backward, 2);
6061 assert_eq!(
6062 backward.messages,
6063 vec![
6064 "row-6", "row-5", "row-4", "row-3", "row-2", "row-1", "row-0"
6065 ]
6066 );
6067 assert_unique_messages(&backward.messages);
6068 assert_eq!(
6069 backward.timestamps,
6070 vec![
6071 start_realtime_usec + 6,
6072 start_realtime_usec + 5,
6073 start_realtime_usec + 4,
6074 start_realtime_usec + 3,
6075 start_realtime_usec + 2,
6076 start_realtime_usec + 1,
6077 start_realtime_usec,
6078 ]
6079 );
6080
6081 let forward = collect_netdata_pages(dir.path(), Direction::Forward, 2);
6082 assert_eq!(
6083 forward.messages,
6084 vec![
6085 "row-1", "row-0", "row-3", "row-2", "row-5", "row-4", "row-6"
6086 ]
6087 );
6088 assert_unique_messages(&forward.messages);
6089 assert_eq!(
6090 forward.timestamps,
6091 vec![
6092 start_realtime_usec + 1,
6093 start_realtime_usec,
6094 start_realtime_usec + 3,
6095 start_realtime_usec + 2,
6096 start_realtime_usec + 5,
6097 start_realtime_usec + 4,
6098 start_realtime_usec + 6,
6099 ]
6100 );
6101 }
6102
6103 #[test]
6104 fn netdata_function_boundary_expansion_does_not_report_more_rows() {
6105 let dir = TempDir::new().expect("tempdir");
6106 let boundary_realtime_usec = 1_700_000_000_000_000;
6107 for name in ["a.journal", "b.journal", "c.journal"] {
6108 write_named_netdata_test_journal(dir.path(), name, 1, boundary_realtime_usec);
6109 }
6110
6111 for direction in [Direction::Backward, Direction::Forward] {
6112 let response = run_netdata_contract_request(
6113 dir.path(),
6114 json!({
6115 "after": 1_700_000_000,
6116 "before": 1_700_000_001,
6117 "last": 2,
6118 "direction": direction_name(direction),
6119 "facets": ["PRIORITY"],
6120 "histogram": "PRIORITY"
6121 }),
6122 );
6123 assert_eq!(response["status"], 200);
6124 assert_eq!(response_column_u64s(&response, "timestamp").len(), 3);
6125 assert_eq!(response["items"]["returned"], 3);
6126 assert_eq!(response["items"]["max_to_return"], 2);
6127 assert_eq!(response["items"]["after"], 0);
6128 assert_eq!(response["items"]["before"], 0);
6129 }
6130 }
6131
6132 #[test]
6133 fn netdata_function_tail_polls_return_only_rows_after_anchor_then_304() {
6134 let dir = TempDir::new().expect("tempdir");
6135 let start_realtime_usec = 1_700_000_000_000_000;
6136 write_stepped_netdata_test_journal(
6137 dir.path(),
6138 "tail-contract.journal",
6139 5,
6140 start_realtime_usec,
6141 1,
6142 );
6143 let anchor = start_realtime_usec + 2;
6144
6145 for requested_direction in [Direction::Backward, Direction::Forward] {
6146 let response = run_netdata_contract_request(
6147 dir.path(),
6148 json!({
6149 "after": 1_700_000_000,
6150 "before": 1_700_000_010,
6151 "last": 5,
6152 "direction": direction_name(requested_direction),
6153 "data_only": true,
6154 "tail": true,
6155 "if_modified_since": anchor,
6156 "anchor": anchor,
6157 }),
6158 );
6159 assert_eq!(response["status"], 200);
6160 assert_eq!(response["_request"]["direction"], "backward");
6161 assert_eq!(
6162 response_column_strings(&response, "MESSAGE"),
6163 vec!["row-4", "row-3"]
6164 );
6165 assert_eq!(
6166 response_column_u64s(&response, "timestamp"),
6167 vec![start_realtime_usec + 4, start_realtime_usec + 3]
6168 );
6169 }
6170
6171 let no_change = run_netdata_contract_request(
6172 dir.path(),
6173 json!({
6174 "after": 1_700_000_000,
6175 "before": 1_700_000_010,
6176 "last": 5,
6177 "direction": "backward",
6178 "data_only": true,
6179 "tail": true,
6180 "if_modified_since": start_realtime_usec + 4,
6181 "anchor": start_realtime_usec + 4,
6182 }),
6183 );
6184 assert_eq!(no_change["status"], 304);
6185 assert_eq!(
6186 no_change["errorMessage"],
6187 "No new data since the previous call."
6188 );
6189 }
6190
6191 #[test]
6192 fn netdata_function_tail_delta_reports_exact_incremental_facets_and_histogram() {
6193 let dir = TempDir::new().expect("tempdir");
6194 let start_realtime_usec = 1_700_000_000_000_000;
6195 write_stepped_netdata_test_journal(
6196 dir.path(),
6197 "delta-contract.journal",
6198 5,
6199 start_realtime_usec,
6200 1,
6201 );
6202 let anchor = start_realtime_usec + 1;
6203
6204 let response = run_netdata_contract_request(
6205 dir.path(),
6206 json!({
6207 "after": 1_700_000_000,
6208 "before": 1_700_000_010,
6209 "last": 2,
6210 "direction": "backward",
6211 "data_only": true,
6212 "delta": true,
6213 "tail": true,
6214 "if_modified_since": anchor,
6215 "anchor": anchor,
6216 "facets": ["SERVICE"],
6217 "histogram": "SERVICE",
6218 }),
6219 );
6220
6221 assert_eq!(response["status"], 200);
6222 assert_eq!(
6223 response_column_strings(&response, "MESSAGE"),
6224 vec!["row-4", "row-3"]
6225 );
6226 assert_eq!(
6227 response_facet_count(&response, "facets_delta", "SERVICE", "even"),
6228 2
6229 );
6230 assert_eq!(
6231 response_facet_count(&response, "facets_delta", "SERVICE", "odd"),
6232 1
6233 );
6234 assert_eq!(
6235 response_histogram_total(&response, "histogram_delta", "even"),
6236 2
6237 );
6238 assert_eq!(
6239 response_histogram_total(&response, "histogram_delta", "odd"),
6240 1
6241 );
6242 let items = response["items_delta"].as_object().expect("items_delta");
6243 assert_eq!(items["matched"], 3);
6244 assert_eq!(items["returned"], 2);
6245 assert_eq!(items["after"], 2);
6246 }
6247
6248 #[test]
6249 fn systemd_profile_keeps_user_group_ids_raw_by_default() {
6250 let context = DisplayContext::default();
6251 let profile = SystemdJournalProfile;
6252 assert_eq!(
6253 profile.field_display_value(&context, DisplayScope::Facet, "_UID", b"0"),
6254 json!("0")
6255 );
6256 assert_eq!(
6257 profile.field_display_value(&context, DisplayScope::Facet, "_GID", b"0"),
6258 json!("0")
6259 );
6260 }
6261
6262 #[cfg(unix)]
6263 #[test]
6264 fn plugin_compatible_profile_resolves_user_group_ids_explicitly() {
6265 let context = DisplayContext::default();
6266 let profile = SystemdJournalPluginProfile;
6267 assert_eq!(
6268 profile.field_display_value(&context, DisplayScope::Facet, "_UID", b"0"),
6269 json!("root")
6270 );
6271 assert_eq!(
6272 profile.field_display_value(&context, DisplayScope::Facet, "_GID", b"0"),
6273 json!("root")
6274 );
6275 }
6276
6277 #[cfg(unix)]
6278 #[test]
6279 fn plugin_compatible_profile_caches_user_group_resolution() {
6280 let context = DisplayContext::default();
6281 let profile = SystemdJournalPluginProfile;
6282
6283 assert_eq!(
6284 profile.field_display_value(&context, DisplayScope::Facet, "_UID", b"0"),
6285 json!("root")
6286 );
6287 assert_eq!(
6288 profile.field_display_value(&context, DisplayScope::Data, "_UID", b"0"),
6289 json!("root")
6290 );
6291 assert_eq!(
6292 profile.field_display_value(&context, DisplayScope::Facet, "_GID", b"0"),
6293 json!("root")
6294 );
6295 assert_eq!(
6296 profile.field_display_value(&context, DisplayScope::Data, "_GID", b"0"),
6297 json!("root")
6298 );
6299
6300 assert_eq!(context.uid_display_cache.borrow().len(), 1);
6301 assert_eq!(context.gid_display_cache.borrow().len(), 1);
6302 }
6303
6304 #[test]
6305 fn file_overlap_uses_netdata_max_realtime_slack() {
6306 let file_first_seconds = 200_000_000u64;
6307 let file_last_seconds = 200_000_100u64;
6308 let slack_seconds = NETDATA_JOURNAL_VS_REALTIME_DELTA_MAX_USEC / 1_000_000;
6309 let header = crate::FileHeader {
6310 signature: *b"LPKSHHRH",
6311 compatible_flags: 0,
6312 incompatible_flags: 0,
6313 state: 0,
6314 file_id: [0; 16],
6315 machine_id: [0; 16],
6316 header_size: 0,
6317 arena_size: 0,
6318 data_hash_table_size: 0,
6319 field_hash_table_size: 0,
6320 n_objects: 0,
6321 n_entries: 0,
6322 head_entry_realtime: file_first_seconds * 1_000_000,
6323 tail_entry_realtime: file_last_seconds * 1_000_000,
6324 tail_entry_monotonic: 0,
6325 head_entry_seqnum: 0,
6326 tail_entry_seqnum: 0,
6327 tail_entry_boot_id: [0; 16],
6328 seqnum_id: [0; 16],
6329 n_data: 0,
6330 n_fields: 0,
6331 n_tags: 0,
6332 n_entry_arrays: 0,
6333 data_hash_chain_depth: 0,
6334 field_hash_chain_depth: 0,
6335 };
6336 let config = NetdataFunctionConfig::systemd_journal();
6337
6338 let inside_slack = NetdataRequest::parse(
6339 &json!({
6340 "after": file_last_seconds + slack_seconds - 1,
6341 "before": file_last_seconds + slack_seconds + 500
6342 }),
6343 &config,
6344 )
6345 .expect("parse request");
6346 assert!(file_may_overlap_request(header, &inside_slack));
6347
6348 let outside_slack = NetdataRequest::parse(
6349 &json!({
6350 "after": file_last_seconds + slack_seconds + 1,
6351 "before": file_last_seconds + slack_seconds + 500
6352 }),
6353 &config,
6354 )
6355 .expect("parse request");
6356 assert!(!file_may_overlap_request(header, &outside_slack));
6357 }
6358
6359 #[test]
6360 fn journal_file_order_matches_plugin_comparator_shape() {
6361 let older = JournalFileOrderInfo {
6362 msg_first_realtime_usec: 100,
6363 msg_last_realtime_usec: 200,
6364 file_last_modified_usec: 200,
6365 journal_vs_realtime_delta_usec: NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC,
6366 };
6367 let newer = JournalFileOrderInfo {
6368 msg_first_realtime_usec: 100,
6369 msg_last_realtime_usec: 300,
6370 file_last_modified_usec: 100,
6371 journal_vs_realtime_delta_usec: NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC,
6372 };
6373 assert_eq!(
6374 compare_journal_file_order(&newer, &older, Direction::Backward),
6375 Ordering::Less
6376 );
6377 assert_eq!(
6378 compare_journal_file_order(&newer, &older, Direction::Forward),
6379 Ordering::Greater
6380 );
6381
6382 let newer_mtime = JournalFileOrderInfo {
6383 msg_first_realtime_usec: 100,
6384 msg_last_realtime_usec: 200,
6385 file_last_modified_usec: 300,
6386 journal_vs_realtime_delta_usec: NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC,
6387 };
6388 assert_eq!(
6389 compare_journal_file_order(&newer_mtime, &older, Direction::Backward),
6390 Ordering::Less
6391 );
6392
6393 let newer_first = JournalFileOrderInfo {
6394 msg_first_realtime_usec: 150,
6395 msg_last_realtime_usec: 200,
6396 file_last_modified_usec: 200,
6397 journal_vs_realtime_delta_usec: NETDATA_JOURNAL_VS_REALTIME_DELTA_DEFAULT_USEC,
6398 };
6399 assert_eq!(
6400 compare_journal_file_order(&newer_first, &older, Direction::Backward),
6401 Ordering::Less
6402 );
6403 }
6404
6405 #[test]
6406 fn boot_first_realtime_keeps_earliest_timestamp_like_plugin() {
6407 let mut boot_first = BTreeMap::new();
6408 record_boot_first_realtime(&mut boot_first, b"boot-a".to_vec(), 300);
6409 record_boot_first_realtime(&mut boot_first, b"boot-a".to_vec(), 100);
6410 record_boot_first_realtime(&mut boot_first, b"boot-a".to_vec(), 200);
6411
6412 assert_eq!(boot_first.get(b"boot-a".as_slice()), Some(&100));
6413 }
6414
6415 #[test]
6416 fn merge_histogram_rejects_inconsistent_bucket_shape() {
6417 let mut target = Some(ExplorerHistogram {
6418 field: b"PRIORITY".to_vec(),
6419 buckets: vec![ExplorerHistogramBucket {
6420 start_realtime_usec: 1_000_000,
6421 end_realtime_usec: 2_000_000,
6422 values: HashMap::new(),
6423 }],
6424 });
6425 let source = ExplorerHistogram {
6426 field: b"PRIORITY".to_vec(),
6427 buckets: vec![ExplorerHistogramBucket {
6428 start_realtime_usec: 1_000_000,
6429 end_realtime_usec: 1_500_000,
6430 values: HashMap::new(),
6431 }],
6432 };
6433
6434 let err = merge_histogram(&mut target, source).expect_err("shape mismatch");
6435 assert!(matches!(
6436 err,
6437 SdkError::Unsupported("inconsistent Netdata histogram bucket shape")
6438 ));
6439 }
6440
6441 #[test]
6442 fn collect_journal_files_recurses_nested_directories() {
6443 let dir = TempDir::new().expect("tempdir");
6444 let nested = dir.path().join("machine").join("nested");
6445 write_named_netdata_test_journal(&nested, "system.journal", 1, 1_700_000_000_000_000);
6446
6447 let collection = collect_journal_files(dir.path()).expect("collect files");
6448
6449 assert_eq!(collection.files.len(), 1);
6450 assert_eq!(collection.skipped, 0);
6451 assert!(collection.errors.is_empty());
6452 assert_eq!(
6453 collection.files[0]
6454 .file_name()
6455 .and_then(|name| name.to_str()),
6456 Some("system.journal")
6457 );
6458 }
6459
6460 #[cfg(unix)]
6461 #[test]
6462 fn collect_journal_files_deduplicates_symlinked_directories() {
6463 let dir = TempDir::new().expect("tempdir");
6464 let real = dir.path().join("real");
6465 let link = dir.path().join("link");
6466 write_named_netdata_test_journal(&real, "system.journal", 1, 1_700_000_000_000_000);
6467 std::os::unix::fs::symlink(&real, &link).expect("symlink");
6468
6469 let collection = collect_journal_files(dir.path()).expect("collect files");
6470
6471 assert_eq!(collection.files.len(), 1);
6472 assert_eq!(collection.skipped, 0);
6473 assert!(collection.errors.is_empty());
6474 }
6475
6476 #[cfg(unix)]
6477 #[test]
6478 fn collect_journal_files_deduplicates_symlinked_files() {
6479 let dir = TempDir::new().expect("tempdir");
6480 write_named_netdata_test_journal(dir.path(), "system.journal", 1, 1_700_000_000_000_000);
6481 std::os::unix::fs::symlink(
6482 dir.path().join("system.journal"),
6483 dir.path().join("system-copy.journal"),
6484 )
6485 .expect("symlink");
6486
6487 let collection = collect_journal_files(dir.path()).expect("collect files");
6488
6489 assert_eq!(collection.files.len(), 1);
6490 assert_eq!(collection.skipped, 0);
6491 assert!(collection.errors.is_empty());
6492 }
6493
6494 #[cfg(unix)]
6495 #[test]
6496 fn collect_journal_files_reports_unreadable_subdirectories() {
6497 use std::os::unix::fs::PermissionsExt;
6498
6499 if unsafe { libc::geteuid() } == 0 {
6500 return;
6501 }
6502
6503 let dir = TempDir::new().expect("tempdir");
6504 std::fs::write(dir.path().join("visible.journal"), b"").expect("journal");
6505 let locked = dir.path().join("locked");
6506 std::fs::create_dir(&locked).expect("locked dir");
6507 std::fs::set_permissions(&locked, std::fs::Permissions::from_mode(0o000))
6508 .expect("lock dir");
6509
6510 let collection = collect_journal_files(dir.path()).expect("collect files");
6511
6512 std::fs::set_permissions(&locked, std::fs::Permissions::from_mode(0o700))
6513 .expect("unlock dir");
6514 assert_eq!(collection.files.len(), 1);
6515 assert_eq!(collection.skipped, 1);
6516 assert_eq!(collection.errors.len(), 1);
6517 assert!(collection.errors[0].contains("locked"));
6518 }
6519
6520 #[test]
6521 fn source_summary_fills_missing_caller_metadata_from_header() {
6522 let dir = TempDir::new().expect("tempdir");
6523 write_named_netdata_test_journal(dir.path(), "system.journal", 2, 1_700_000_000_000_000);
6524 let path = dir.path().join("system.journal");
6525 let mut state = TestNetdataState::default();
6526 state.metadata.insert(
6527 path.clone(),
6528 NetdataJournalFileMetadata {
6529 msg_first_realtime_usec: Some(1_699_999_999_000_000),
6530 ..NetdataJournalFileMetadata::default()
6531 },
6532 );
6533 let options = NetdataFunctionRunOptions {
6534 state: Some(&mut state),
6535 ..NetdataFunctionRunOptions::default()
6536 };
6537
6538 let summary = JournalSourceSummary::from_paths(&[path], ReaderOptions::default(), &options);
6539
6540 assert_eq!(summary.first_realtime_usec, Some(1_699_999_999_000_000));
6541 assert_eq!(summary.last_realtime_usec, Some(1_700_000_000_000_001));
6542 }
6543
6544 #[test]
6545 fn source_summary_coverage_is_off_below_one_second() {
6546 let first = 1_700_000_000_000_000;
6547 let mut summary = JournalSourceSummary {
6548 files: 1,
6549 total_size: 0,
6550 first_realtime_usec: Some(first),
6551 last_realtime_usec: Some(first + 999_999),
6552 };
6553
6554 assert!(summary.info().contains("covering off"));
6555
6556 summary.last_realtime_usec = Some(first + 1_000_000);
6557 assert!(summary.info().contains("covering 1s"));
6558 }
6559
6560 #[test]
6561 fn netdata_function_uses_default_source_selector_metadata() {
6562 let dir = TempDir::new().expect("tempdir");
6563 let response = NetdataJournalFunction::systemd_journal_plugin_compatible()
6564 .run_directory_request_json(dir.path(), &json!({"info": true}))
6565 .expect("run info request");
6566 let source = response["required_params"]
6567 .as_array()
6568 .expect("required params")
6569 .iter()
6570 .find(|param| param["id"] == "__logs_sources")
6571 .expect("source selector param");
6572
6573 assert_eq!(source["id"], "__logs_sources");
6574 assert_eq!(source["name"], DEFAULT_SOURCE_SELECTOR_NAME);
6575 assert_eq!(source["help"], DEFAULT_SOURCE_SELECTOR_HELP);
6576 }
6577
6578 #[test]
6579 fn netdata_function_uses_custom_source_selector_metadata() {
6580 let dir = TempDir::new().expect("tempdir");
6581 let mut config = NetdataFunctionConfig::systemd_journal();
6582 config.source_selector_name = "Trap Jobs".to_string();
6583 config.source_selector_help = "Select the trap job to query".to_string();
6584 let function = NetdataJournalFunction::new(config, SystemdJournalPluginProfile);
6585
6586 let response = function
6587 .run_directory_request_json(dir.path(), &json!({"info": true}))
6588 .expect("run info request");
6589 let source = response["required_params"]
6590 .as_array()
6591 .expect("required params")
6592 .iter()
6593 .find(|param| param["id"] == "__logs_sources")
6594 .expect("source selector param");
6595
6596 assert_eq!(source["id"], "__logs_sources");
6597 assert_eq!(source["name"], "Trap Jobs");
6598 assert_eq!(source["help"], "Select the trap job to query");
6599 }
6600
6601 #[test]
6602 fn netdata_function_empty_source_selector_metadata_falls_back_to_defaults() {
6603 let dir = TempDir::new().expect("tempdir");
6604 let mut config = NetdataFunctionConfig::systemd_journal();
6605 config.source_selector_name.clear();
6606 config.source_selector_help.clear();
6607 let function = NetdataJournalFunction::new(config, SystemdJournalPluginProfile);
6608
6609 let response = function
6610 .run_directory_request_json(dir.path(), &json!({"info": true}))
6611 .expect("run info request");
6612 let source = response["required_params"]
6613 .as_array()
6614 .expect("required params")
6615 .iter()
6616 .find(|param| param["id"] == "__logs_sources")
6617 .expect("source selector param");
6618
6619 assert_eq!(source["id"], "__logs_sources");
6620 assert_eq!(source["name"], DEFAULT_SOURCE_SELECTOR_NAME);
6621 assert_eq!(source["help"], DEFAULT_SOURCE_SELECTOR_HELP);
6622 }
6623
6624 #[test]
6625 fn source_selection_echoes_and_filters_known_groups() {
6626 let config = NetdataFunctionConfig::systemd_journal();
6627 let request = NetdataRequest::parse(
6628 &json!({
6629 "selections": {
6630 "__logs_sources": ["all-local-system-logs"]
6631 }
6632 }),
6633 &config,
6634 )
6635 .expect("parse source-filtered request");
6636
6637 assert_eq!(request.source_type, SOURCE_TYPE_LOCAL_SYSTEM);
6638 assert_eq!(
6639 request.echo.get("source_type").and_then(Value::as_u64),
6640 Some(SOURCE_TYPE_LOCAL_SYSTEM)
6641 );
6642 assert!(
6643 request
6644 .echo
6645 .pointer("/selections/__logs_sources/0")
6646 .is_some_and(Value::is_null)
6647 );
6648 assert!(request.matches_source(Path::new("/var/log/journal/machine/system.journal"), None));
6649 assert!(!request.matches_source(
6650 Path::new("/var/log/journal/machine/user-1000.journal"),
6651 None
6652 ));
6653 }
6654
6655 #[test]
6656 fn source_selection_uses_caller_metadata_before_filename_fallback() {
6657 let dir = TempDir::new().expect("tempdir");
6658 write_named_netdata_test_journal(dir.path(), "user-1000.journal", 1, 1_700_000_000_000_000);
6659 let path = dir.path().join("user-1000.journal");
6660 let config = NetdataFunctionConfig::systemd_journal();
6661 let request = NetdataRequest::parse(
6662 &json!({
6663 "after": 1_700_000_000,
6664 "before": 1_700_000_001,
6665 "selections": {
6666 "__logs_sources": ["all-local-system-logs"]
6667 }
6668 }),
6669 &config,
6670 )
6671 .expect("parse source-filtered request");
6672 assert!(!request.matches_source(&path, None));
6673
6674 let mut state = TestNetdataState::default();
6675 state.metadata.insert(
6676 path.clone(),
6677 NetdataJournalFileMetadata {
6678 source_type: Some(NETDATA_SOURCE_TYPE_LOCAL_SYSTEM),
6679 source_name: Some("system-registry".to_string()),
6680 ..NetdataJournalFileMetadata::default()
6681 },
6682 );
6683 let options = NetdataFunctionRunOptions {
6684 state: Some(&mut state),
6685 ..NetdataFunctionRunOptions::default()
6686 };
6687 let selected =
6688 select_journal_files_for_request(vec![path], &request, config.reader_options, &options);
6689
6690 assert_eq!(selected.files.len(), 1);
6691 }
6692
6693 #[test]
6694 fn netdata_function_state_receives_learned_source_realtime_delta() {
6695 let dir = TempDir::new().expect("tempdir");
6696 let commit_realtime_usec = 1_700_000_030_000_000;
6697 let source_realtime_usec = commit_realtime_usec - 30_000_000;
6698 write_source_realtime_delta_journal(
6699 dir.path(),
6700 "system.journal",
6701 commit_realtime_usec,
6702 source_realtime_usec,
6703 );
6704 let request = json!({
6705 "after": 1_700_000_029,
6706 "before": 1_700_000_031,
6707 "facets": ["SERVICE"],
6708 "histogram": "SERVICE",
6709 "last": 1,
6710 "sampling": 0
6711 });
6712 let function = NetdataJournalFunction::systemd_journal_plugin_compatible();
6713 let mut state = TestNetdataState::default();
6714 let options = NetdataFunctionRunOptions {
6715 state: Some(&mut state),
6716 ..NetdataFunctionRunOptions::from_timeout_seconds(0)
6717 };
6718
6719 let response = function
6720 .run_directory_request_json_with_options(dir.path(), &request, options)
6721 .expect("run function");
6722
6723 assert_eq!(response["status"], 200);
6724 assert_eq!(state.updates.len(), 1);
6725 assert_eq!(state.updates[0].1, 30_000_000);
6726 }
6727
6728 #[test]
6729 fn source_classification_matches_plugin_filename_shape() {
6730 assert_eq!(
6731 journal_file_source_type(Path::new("/var/log/journal/machine/system.journal")),
6732 SOURCE_TYPE_ALL | SOURCE_TYPE_LOCAL_ALL | SOURCE_TYPE_LOCAL_SYSTEM
6733 );
6734 assert_eq!(
6735 journal_file_source_type(Path::new("/var/log/journal/machine/user-1000.journal")),
6736 SOURCE_TYPE_ALL | SOURCE_TYPE_LOCAL_ALL | SOURCE_TYPE_LOCAL_USER
6737 );
6738 assert_eq!(
6739 journal_file_source_type(Path::new("/var/log/journal/machine/other.journal")),
6740 SOURCE_TYPE_ALL | SOURCE_TYPE_LOCAL_ALL | SOURCE_TYPE_LOCAL_OTHER
6741 );
6742 assert_eq!(
6743 journal_file_source_type(Path::new(
6744 "/var/log/journal/machine.namespace/system.journal"
6745 )),
6746 SOURCE_TYPE_ALL | SOURCE_TYPE_LOCAL_ALL | SOURCE_TYPE_LOCAL_NAMESPACE
6747 );
6748 assert_eq!(
6749 journal_file_source_type(Path::new(
6750 "/var/log/journal/remote/remote-host-a@machine.journal"
6751 )),
6752 SOURCE_TYPE_ALL | SOURCE_TYPE_REMOTE_ALL
6753 );
6754 }
6755
6756 #[test]
6757 fn exact_source_names_follow_plugin_prefixes() {
6758 assert_eq!(
6759 journal_file_exact_source_name(Path::new(
6760 "/var/log/journal/machine.namespace/system.journal"
6761 ))
6762 .as_deref(),
6763 Some("namespace-namespace")
6764 );
6765 assert_eq!(
6766 journal_file_exact_source_name(Path::new(
6767 "/var/log/journal/remote/remote-host-a@machine.journal"
6768 ))
6769 .as_deref(),
6770 Some("remote-host-a")
6771 );
6772 assert_eq!(
6773 journal_file_exact_source_name(Path::new(
6774 "/var/log/journal/remote/remote-host-b.journal~.zst"
6775 ))
6776 .as_deref(),
6777 Some("remote-host-b")
6778 );
6779 }
6780
6781 #[test]
6782 fn disposed_journal_extension_matches_plugin_scan_contract() {
6783 assert!(is_journal_file_name(Path::new("active.journal")));
6784 assert!(is_journal_file_name(Path::new("archived.journal~")));
6785 assert!(is_journal_file_name(Path::new("active.journal.zst")));
6786 assert!(is_journal_file_name(Path::new("archived.journal~.zst")));
6787 }
6788
6789 fn test_located_row(realtime_usec: u64) -> LocatedRow {
6790 LocatedRow {
6791 file_path: PathBuf::from("test.journal"),
6792 row: ExplorerRow {
6793 realtime_usec,
6794 cursor: String::new(),
6795 payloads: Vec::new(),
6796 },
6797 }
6798 }
6799
6800 fn test_located_row_in_file(file_path: &str, realtime_usec: u64, cursor: &str) -> LocatedRow {
6801 LocatedRow {
6802 file_path: PathBuf::from(file_path),
6803 row: ExplorerRow {
6804 realtime_usec,
6805 cursor: cursor.to_string(),
6806 payloads: Vec::new(),
6807 },
6808 }
6809 }
6810
6811 fn write_source_realtime_delta_journal(
6812 directory: &std::path::Path,
6813 name: &str,
6814 commit_realtime_usec: u64,
6815 source_realtime_usec: u64,
6816 ) {
6817 std::fs::create_dir_all(directory).expect("create journal dir");
6818 let path = directory.join(name);
6819 let repo_file = RepoFile::from_path(&path).expect("repo file");
6820 let options = JournalFileOptions::new(test_uuid(1), test_uuid(2), test_uuid(3));
6821 let mut file = JournalFile::<MmapMut>::create(&repo_file, options).expect("create journal");
6822 let mut writer = JournalWriter::new(&mut file, 1, test_uuid(4)).expect("writer");
6823 let source = format!("_SOURCE_REALTIME_TIMESTAMP={source_realtime_usec}");
6824 let payloads: [&[u8]; 4] = [
6825 b"MESSAGE=source lag test".as_slice(),
6826 b"SERVICE=delta".as_slice(),
6827 b"PRIORITY=6".as_slice(),
6828 source.as_bytes(),
6829 ];
6830 writer
6831 .add_entry(
6832 &mut file,
6833 &payloads,
6834 commit_realtime_usec,
6835 commit_realtime_usec,
6836 )
6837 .expect("write entry");
6838 file.sync().expect("sync journal");
6839 }
6840}