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