@cryptotaxi247 / netdata-1 / commits / 5c2b637c5

Exclude ND_REMAPPING bookkeeping entries from journal index (#22513)

* Exclude ND_REMAPPING bookkeeping entries from journal index * Propagate errors when reading remapping field info

vkalintiris committed May 21, 2026 at 11:48 UTC 5c2b637c5ed4379f621a82d63a290872e4746074

2 files changed +111 -8

src/crates/journal-engine/src/cache.rs

+7 -1

@@ -8,7 +8,13 @@ use serde::{Deserialize, Serialize};

9 /// Cache version number. Increment this when the FileIndex or FileIndexKey

10 /// schema changes to automatically invalidate old cache entries.

11 -const CACHE_VERSION: u32 = 1;

11 +///

12 +/// v2: ND_REMAPPING bookkeeping entries are now excluded from the index's

13 +/// time histogram, time-ordered entry list, and per-field bitmaps. Old

14 +/// v1 caches contain inflated `total_entries` and a phantom "(unset)"

15 +/// contribution in the histogram bucket holding the remapping entry's

16 +/// `__REALTIME_TIMESTAMP`.

17 +const CACHE_VERSION: u32 = 2;

19 /// Cache key for file indexes that includes the file, facets, source timestamp

20 /// field, and cache version. Different facet configurations or timestamp fields

src/crates/journal-index/src/file_indexer.rs

+104 -7

@@ -12,6 +12,7 @@ use crate::{

12 Seconds,

13 };

14 use journal_core::collections::{HashMap, HashSet};

15 +use journal_core::field_map::REMAPPING_MARKER;

16 use journal_core::file::{JournalFile, Mmap, offset_array::InlinedCursor};

17 use journal_registry::File;

18 use std::num::NonZeroU64;

@@ -101,6 +102,14 @@ pub struct FileIndexer {

102 // Maps entry offsets to an index of an implicitly defined time-ordered

103 // array of entries

104 entry_offset_index: HashMap<NonZeroU64, u64>,

105 +

106 + // Entry offsets that belong to ND_REMAPPING=1 bookkeeping records written

107 + // by the otel-plugin. These describe the OTel-to-systemd field name

108 + // mapping; they are not real log records and must be excluded from the

109 + // file's time histogram, the time-ordered entry list, and the per-field

110 + // bitmaps so that downstream consumers (queries, histograms) see only

111 + // genuine log entries.

112 + remapping_entry_offsets: HashSet<NonZeroU64>,

113 }

114

115 impl Default for FileIndexer {

@@ -120,6 +129,7 @@ impl FileIndexer {

129 realtime_entry_offset_pairs: Vec::new(),

130 entry_indices: Vec::new(),

131 entry_offset_index: HashMap::default(),

132 + remapping_entry_offsets: HashSet::default(),

133 }

134 }

135 }

@@ -145,6 +155,7 @@ impl FileIndexer {

155 self.entry_indices = Vec::new();

156 self.entry_offsets = Vec::new();

157 self.entry_offset_index = HashMap::default();

158 + self.remapping_entry_offsets.clear();

159

160 let window_size = 32 * 1024 * 1024;

161 let journal_file = JournalFile::<Mmap>::open(file, window_size)?;

@@ -186,6 +197,13 @@ impl FileIndexer {

197

198 let field_map = journal_file.load_fields()?;

199

200 + // Discover ND_REMAPPING bookkeeping entries so they can be excluded

201 + // from the time histogram, the time-ordered entry list, and the

202 + // per-field bitmaps. Typically there is at most one such entry per

203 + // file (written by the otel-plugin); journals without remappings

204 + // produce an empty set.

205 + self.collect_remapping_entry_offsets(&journal_file, tail_object_offset)?;

206 +

207 // Build the file histogram

208 let histogram = self.build_histogram(

209 &journal_file,

@@ -300,11 +318,6 @@ impl FileIndexer {

318 continue;

319 }

320

303 - // Skip the remapping value

304 - if data_object.raw_payload().ends_with(field_name.as_bytes()) {

305 - continue;

306 - };

307 -

321 let data_payload =

322 String::from_utf8_lossy(data_object.raw_payload()).into_owned();

323 let Some(inlined_cursor) = data_object.inlined_cursor() else {

@@ -330,7 +343,9 @@ impl FileIndexer {

343 }

344

345 // Map entry offsets where this data object appears to entry indices.

333 - // Filter out any offsets that are beyond our initial snapshot's maximum

346 + // Filter out any offsets that are beyond our initial snapshot's maximum,

347 + // and any offsets that belong to ND_REMAPPING bookkeeping records (those

348 + // entries are not in entry_offset_index and must not appear in bitmaps).

349 self.entry_indices.clear();

350 for entry_offset in self

351 .entry_offsets

@@ -338,8 +353,12 @@ impl FileIndexer {

353 .copied()

354 .filter(|offset| *offset <= tail_object_offset)

355 {

356 + if self.remapping_entry_offsets.contains(&entry_offset) {

357 + continue;

358 + }

359 let Some(entry_index) = self.entry_offset_index.get(&entry_offset) else {

342 - // This should never happen given that we filter by the tail object offset.

360 + // This should never happen given that we filter by the tail object

361 + // offset and exclude remapping entries.

362 panic!(

363 "missing entry offset {} from index (total offsets: {})",

364 entry_offset,

@@ -348,6 +367,14 @@ impl FileIndexer {

367 };

368 self.entry_indices.push(*entry_index as u32);

369 }

370 +

371 + // If every entry that contains this data object is a remapping

372 + // record, the data object only describes the OTel field mapping

373 + // (e.g. NDABE_LOG_SEVERITY_NUMBER=log.severity_number) and must

374 + // not surface as a value in the index.

375 + if self.entry_indices.is_empty() {

376 + continue;

377 + }

378 self.entry_indices.sort_unstable();

379

380 // Create the bitmap for the entry indices

@@ -408,6 +435,72 @@ impl FileIndexer {

435 Ok(entries_index)

436 }

437

438 + /// Collect entry offsets that belong to ND_REMAPPING=1 bookkeeping

439 + /// records.

440 + ///

441 + /// The otel-plugin writes one such entry per journal file containing the

442 + /// mapping from OTel field names to their systemd-compatible counterparts

443 + /// (e.g. `NDABE_LOG_SEVERITY_NUMBER=log.severity_number`). These records

444 + /// are not log messages and must be excluded from time histograms,

445 + /// time-ordered entry lists, and per-field bitmaps. Journals that contain

446 + /// no remappings (e.g. regular systemd journals) leave the set empty.

447 + ///

448 + /// Only entries whose offset is at or before `tail_object_offset` are

449 + /// considered, matching the snapshot semantics used elsewhere in

450 + /// indexing.

451 + fn collect_remapping_entry_offsets(

452 + &mut self,

453 + journal_file: &JournalFile<Mmap>,

454 + tail_object_offset: NonZeroU64,

455 + ) -> Result<()> {

456 + // The ND_REMAPPING field key is the bytes before '=' in the marker.

457 + // Resolved from REMAPPING_MARKER to keep both call sites in sync.

458 + let Some(eq_pos) = REMAPPING_MARKER.iter().position(|b| *b == b'=') else {

459 + // REMAPPING_MARKER is a compile-time constant containing '='; this

460 + // branch is unreachable but is preferred over an unwrap so changes

461 + // to the marker can't crash indexing.

462 + return Ok(());

463 + };

464 + let marker_field = &REMAPPING_MARKER[..eq_pos];

465 +

466 + // `field_data_objects` returns an empty iterator when the field is

467 + // absent (the regular-systemd-journal case), so any Err here is a

468 + // genuine journal read failure. Swallowing it would leave the

469 + // indexer unaware of bookkeeping records and re-introduce the leak

470 + // this helper exists to prevent; propagate instead. The same

471 + // reasoning applies to the per-data-object and per-cursor errors

472 + // below — a missed ND_REMAPPING data object means its entries

473 + // silently flow into the index.

474 + let field_data_iterator = journal_file.field_data_objects(marker_field)?;

475 +

476 + for data_object in field_data_iterator {

477 + let data_object = data_object?;

478 +

479 + // Only the exact `ND_REMAPPING=1` payload marks a remapping entry.

480 + if data_object.raw_payload() != REMAPPING_MARKER {

481 + continue;

482 + }

483 +

484 + let Some(inlined_cursor) = data_object.inlined_cursor() else {

485 + continue;

486 + };

487 +

488 + self.entry_offsets.clear();

489 + inlined_cursor.collect_offsets(journal_file, &mut self.entry_offsets)?;

490 +

491 + for entry_offset in self

492 + .entry_offsets

493 + .iter()

494 + .copied()

495 + .filter(|offset| *offset <= tail_object_offset)

496 + {

497 + self.remapping_entry_offsets.insert(entry_offset);

498 + }

499 + }

500 +

501 + Ok(())

502 + }

503 +

504 /// Collect timestamp information from a source timestamp field.

505 ///

506 /// This extracts (timestamp, entry_offset) pairs from the specified source

@@ -474,6 +567,9 @@ impl FileIndexer {

567 }

568

569 for entry_offset in &self.entry_offsets {

570 + if self.remapping_entry_offsets.contains(entry_offset) {

571 + continue;

572 + }

573 self.source_timestamp_entry_offset_pairs

574 .push((*ts, *entry_offset));

575 }

@@ -536,6 +632,7 @@ impl FileIndexer {

632 .iter()

633 .copied()

634 .filter(|offset| *offset <= tail_object_offset)

635 + .filter(|offset| !self.remapping_entry_offsets.contains(offset))

636 {

637 if self.entry_offset_index.contains_key(&entry_offset) {

638 // We have the timestamp of this entry offset