Handle fields with high-cardinality and big payloads. (#21716)
* Remove verbose logs and fix log-levels `tracing` maps debug logs to info on systemd. Remove verbose logs and set log level to trace for those we want to keep. * Add configurable journal file indexing limits Add max_unique_values_per_field and max_field_payload_size options to protect against memory exhaustion when indexing high-cardinality fields or large payloads. Limits are configurable via journal-viewer.yaml. * Deduplicate incoming facets.
vkalintiris committed
Feb 6, 2026 at 00:10 UTC
80a5a9f63ea05ce378cf051c809076a21f519557
11 files changed
+240
-34
src/crates/journal-engine/examples/index.rs
+11
-3
@@ -32,7 +32,8 @@
32
33
use foundation::Timeout;
34
use journal_engine::{
35
- Facets, FileIndexCacheBuilder, FileIndexKey, QueryTimeRange, batch_compute_file_indexes,
35
+ Facets, FileIndexCacheBuilder, FileIndexKey, IndexingLimits, QueryTimeRange,
36
+ batch_compute_file_indexes,
37
};
38
use journal_index::FieldName;
39
use journal_registry::{Monitor, Registry};
@@ -120,8 +121,15 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
121
122
// Run batch indexing
123
let start = std::time::Instant::now();
123
- let responses =
124
- batch_compute_file_indexes(&cache, ®istry, keys, &time_range, timeout).await?;
124
+ let responses = batch_compute_file_indexes(
125
+ &cache,
126
+ ®istry,
127
+ keys,
128
+ &time_range,
129
+ timeout,
130
+ IndexingLimits::default(),
131
+ )
132
+ .await?;
133
134
let elapsed = start.elapsed();
135
src/crates/journal-engine/src/facets.rs
+2
-1
@@ -103,8 +103,9 @@ impl Facets {
103
.collect()
104
};
105
106
- // Sort in order to get the same hash for the same set of fields
106
+ // Sort and deduplicate to get a canonical set of fields
107
facets.sort();
108
+ facets.dedup();
109
110
use std::hash::Hasher;
111
let mut hasher = std::hash::DefaultHasher::new();
src/crates/journal-engine/src/indexing.rs
+5
-3
@@ -10,7 +10,7 @@ use crate::{
10
query_time_range::QueryTimeRange,
11
};
12
use foundation::Timeout;
13
-use journal_index::{FileIndex, FileIndexer};
13
+use journal_index::{FileIndex, FileIndexer, IndexingLimits};
14
use journal_registry::Registry;
15
use tracing::{error, trace};
16
@@ -136,8 +136,9 @@ impl Default for FileIndexCacheBuilder {
136
/// * `cache` - The file index cache
137
/// * `registry` - Registry to update with file metadata
138
/// * `keys` - Vector of (file, facets, source_timestamp_field) to fetch/compute indexes for
139
-/// * `bucket_duration` - Duration of histogram buckets in seconds
139
+/// * `time_range` - Query time range for bucket duration calculation
140
/// * `timeout` - Timeout for the entire operation (can be extended dynamically)
141
+/// * `indexing_limits` - Configuration limits for indexing (cardinality, payload size)
142
///
143
/// # Returns
144
/// Vector of responses for each key. Successful responses contain the file index.
@@ -148,6 +149,7 @@ pub async fn batch_compute_file_indexes(
149
keys: Vec<FileIndexKey>,
150
time_range: &QueryTimeRange,
151
timeout: Timeout,
152
+ indexing_limits: IndexingLimits,
153
) -> Result<Vec<(FileIndexKey, FileIndex)>> {
154
let bucket_duration = time_range.bucket_duration_seconds();
155
// Phase 1: Batch check cache for all keys upfront
@@ -240,7 +242,7 @@ pub async fn batch_compute_file_indexes(
242
return (key, Err(EngineError::TimeBudgetExceeded));
243
}
244
243
- let mut file_indexer = FileIndexer::default();
245
+ let mut file_indexer = FileIndexer::new(indexing_limits);
246
let result = file_indexer
247
.index(
248
&key.file,
src/crates/journal-engine/src/lib.rs
+1
@@ -34,5 +34,6 @@ pub use histogram::{
34
BucketRequest, BucketResponse, Histogram, HistogramEngine, calculate_bucket_duration,
35
};
36
pub use indexing::{FileIndexCacheBuilder, batch_compute_file_indexes};
37
+pub use journal_index::IndexingLimits;
38
pub use logs::{CellValue, ColumnInfo, LogEntryData, LogQuery, Table, entry_data_to_table};
39
pub use query_time_range::QueryTimeRange;
src/crates/journal-index/src/file_indexer.rs
+125
-1
@@ -17,6 +17,42 @@ use journal_registry::File;
17
use std::num::NonZeroU64;
18
use tracing::{error, warn};
19
20
+/// Default maximum number of unique values to index per field.
21
+pub const DEFAULT_MAX_UNIQUE_VALUES_PER_FIELD: usize = 500;
22
+
23
+/// Default maximum payload size (in bytes) for field values to index.
24
+pub const DEFAULT_MAX_FIELD_PAYLOAD_SIZE: usize = 100;
25
+
26
+/// Configuration limits for the indexing process.
27
+///
28
+/// These limits protect against unbounded memory growth when indexing
29
+/// journal files with high-cardinality fields or large payloads.
30
+#[derive(Debug, Clone, Copy)]
31
+pub struct IndexingLimits {
32
+ /// Maximum number of unique values to index per field.
33
+ ///
34
+ /// Fields with more unique values than this limit will have their indexing
35
+ /// truncated. This protects against high-cardinality fields (e.g., MESSAGE
36
+ /// with millions of unique values) causing memory exhaustion.
37
+ pub max_unique_values_per_field: usize,
38
+
39
+ /// Maximum payload size (in bytes) for field values to index.
40
+ ///
41
+ /// Field values with payloads larger than this limit (or compressed values)
42
+ /// will be skipped. This prevents large binary data or encoded content
43
+ /// from consuming excessive memory.
44
+ pub max_field_payload_size: usize,
45
+}
46
+
47
+impl Default for IndexingLimits {
48
+ fn default() -> Self {
49
+ Self {
50
+ max_unique_values_per_field: DEFAULT_MAX_UNIQUE_VALUES_PER_FIELD,
51
+ max_field_payload_size: DEFAULT_MAX_FIELD_PAYLOAD_SIZE,
52
+ }
53
+ }
54
+}
55
+
56
/// Reusable indexer for creating searchable indexes from journal files.
57
///
58
/// # Indexing Process
@@ -39,9 +75,12 @@ use tracing::{error, warn};
75
/// The indexer captures the journal file's `tail_object_offset` at the start of indexing
76
/// to create a consistent snapshot. Any entries written to the file after indexing begins
77
/// are ignored, preventing race conditions with concurrent writers.
42
-#[derive(Debug, Default)]
78
+#[derive(Debug)]
79
#[cfg_attr(feature = "allocative", derive(allocative::Allocative))]
80
pub struct FileIndexer {
81
+ /// Configuration limits for the indexing process.
82
+ limits: IndexingLimits,
83
+
84
// Associates a source timestamp value with its inlined cursor
85
source_timestamp_cursor_pairs: Vec<(Microseconds, InlinedCursor)>,
86
@@ -64,8 +103,35 @@ pub struct FileIndexer {
103
entry_offset_index: HashMap<NonZeroU64, u64>,
104
}
105
106
+impl Default for FileIndexer {
107
+ fn default() -> Self {
108
+ Self::new(IndexingLimits::default())
109
+ }
110
+}
111
+
112
+impl FileIndexer {
113
+ /// Create a new indexer with the specified configuration limits.
114
+ pub fn new(limits: IndexingLimits) -> Self {
115
+ Self {
116
+ limits,
117
+ source_timestamp_cursor_pairs: Vec::new(),
118
+ entry_offsets: Vec::new(),
119
+ source_timestamp_entry_offset_pairs: Vec::new(),
120
+ realtime_entry_offset_pairs: Vec::new(),
121
+ entry_indices: Vec::new(),
122
+ entry_offset_index: HashMap::default(),
123
+ }
124
+ }
125
+}
126
+
127
impl FileIndexer {
128
/// Create a searchable index from a journal file.
129
+ ///
130
+ /// # Arguments
131
+ /// * `file` - The journal file to index
132
+ /// * `source_timestamp_field` - Optional field to use for timestamps
133
+ /// * `field_names` - Fields to create bitmap indexes for
134
+ /// * `bucket_duration` - Duration of histogram buckets
135
pub fn index(
136
&mut self,
137
file: &File,
@@ -168,6 +234,9 @@ impl FileIndexer {
234
///
235
/// Only entries with offsets <= `tail_object_offset` are included in the
236
/// bitmaps, ensuring a consistent snapshot.
237
+ ///
238
+ /// Fields with more than `self.limits.max_unique_values_per_field` unique values
239
+ /// will have their indexing truncated to prevent unbounded memory growth.
240
fn build_entries_index(
241
&mut self,
242
journal_file: &JournalFile<Mmap>,
@@ -176,6 +245,8 @@ impl FileIndexer {
245
tail_object_offset: NonZeroU64,
246
) -> Result<HashMap<FieldValuePair, Bitmap>> {
247
let mut entries_index = HashMap::default();
248
+ let mut truncated_fields: Vec<&FieldName> = Vec::new();
249
+ let mut fields_with_large_payloads: Vec<&FieldName> = Vec::new();
250
251
for field_name in field_names {
252
let Some(systemd_field) = field_map.get(field_name.as_str()) else {
@@ -197,13 +268,32 @@ impl FileIndexer {
268
}
269
};
270
271
+ // Track the number of unique values indexed for this field
272
+ let mut unique_values_count: usize = 0;
273
+ let mut ignored_large_payloads: usize = 0;
274
+ let mut was_truncated = false;
275
+
276
for data_object in field_data_iterator {
277
+ // Check cardinality limit before processing this value
278
+ if unique_values_count >= self.limits.max_unique_values_per_field {
279
+ was_truncated = true;
280
+ break;
281
+ }
282
+
283
// Get the payload and the inlined cursor for this data object
284
let (data_payload, inlined_cursor) = {
285
let Ok(data_object) = data_object else {
286
continue;
287
};
288
289
+ // Do not create indexes with fields that contain large payloads.
290
+ if data_object.raw_payload().len() >= self.limits.max_field_payload_size
291
+ || data_object.is_compressed()
292
+ {
293
+ ignored_large_payloads += 1;
294
+ continue;
295
+ }
296
+
297
// Skip the remapping value
298
if data_object.raw_payload().ends_with(field_name.as_bytes()) {
299
continue;
@@ -262,7 +352,41 @@ impl FileIndexer {
352
let field_name = FieldName::new_unchecked(field_name);
353
let k = FieldValuePair::new_unchecked(field_name, String::from(pair.value()));
354
entries_index.insert(k, bitmap);
355
+
356
+ unique_values_count += 1;
357
+ }
358
+
359
+ // Track fields that were truncated or had large payloads skipped
360
+ if was_truncated {
361
+ truncated_fields.push(field_name);
362
}
363
+ if ignored_large_payloads > 0 {
364
+ fields_with_large_payloads.push(field_name);
365
+ }
366
+ }
367
+
368
+ // Log summary of indexing issues
369
+ if !truncated_fields.is_empty() {
370
+ let field_names: Vec<&str> = truncated_fields.iter().map(|f| f.as_str()).collect();
371
+ warn!(
372
+ "File '{}': {} field(s) truncated due to cardinality limit ({}): {:?}",
373
+ journal_file.file().path(),
374
+ truncated_fields.len(),
375
+ self.limits.max_unique_values_per_field,
376
+ field_names
377
+ );
378
+ }
379
+ if !fields_with_large_payloads.is_empty() {
380
+ let field_names: Vec<&str> = fields_with_large_payloads
381
+ .iter()
382
+ .map(|f| f.as_str())
383
+ .collect();
384
+ tracing::info!(
385
+ "File '{}': {} field(s) had values skipped due to large payloads: {:?}",
386
+ journal_file.file().path(),
387
+ fields_with_large_payloads.len(),
388
+ field_names
389
+ );
390
}
391
392
Ok(entries_index)
src/crates/journal-index/src/lib.rs
+3
-1
@@ -20,7 +20,9 @@ pub use file_index::{
20
};
21
22
pub mod file_indexer;
23
-pub use file_indexer::FileIndexer;
23
+pub use file_indexer::{
24
+ FileIndexer, IndexingLimits, DEFAULT_MAX_FIELD_PAYLOAD_SIZE, DEFAULT_MAX_UNIQUE_VALUES_PER_FIELD,
25
+};
26
27
pub mod bitmap;
28
pub use bitmap::Bitmap;
src/crates/netdata-log-viewer/journal-function/src/lib.rs
+3
-3
@@ -11,9 +11,9 @@ pub mod netdata;
11
// Re-export types from journal-engine for convenience
12
pub use journal_engine::{
13
BucketRequest, BucketResponse, CellValue, ColumnInfo, Facets, FileIndexCache,
14
- FileIndexCacheBuilder, FileIndexKey, Histogram, HistogramEngine, LogEntryData, LogQuery,
15
- QueryTimeRange, Result, Table, batch_compute_file_indexes, calculate_bucket_duration,
16
- entry_data_to_table,
14
+ FileIndexCacheBuilder, FileIndexKey, Histogram, HistogramEngine, IndexingLimits,
15
+ LogEntryData, LogQuery, QueryTimeRange, Result, Table, batch_compute_file_indexes,
16
+ calculate_bucket_duration, entry_data_to_table,
17
};
18
19
// Re-export Timeout from foundation (via rt for backward compatibility)
src/crates/netdata-log-viewer/journal-viewer-plugin/configs/journal-viewer.yaml.in
+16
@@ -43,3 +43,19 @@ cache:
43
# Controls backpressure on the indexing system
44
# Default: 100
45
queue_capacity: 100
46
+
47
+indexing:
48
+ # Maximum number of unique values to index per field.
49
+ # Fields with more unique values than this limit will have their indexing
50
+ # truncated to prevent unbounded memory growth. This protects against
51
+ # high-cardinality fields (e.g., MESSAGE with millions of unique values)
52
+ # causing memory exhaustion during indexing.
53
+ # Default: 500
54
+ max_unique_values_per_field: 500
55
+
56
+ # Maximum payload size (in bytes) for field values to index.
57
+ # Field values with payloads larger than this limit (or compressed values)
58
+ # will be skipped. This prevents large binary data or encoded content
59
+ # from consuming excessive memory.
60
+ # Default: 100
61
+ max_field_payload_size: 100
src/crates/netdata-log-viewer/journal-viewer-plugin/src/catalog.rs
+11
-20
@@ -11,8 +11,8 @@ use tracing::{debug, error, info, instrument, warn};
11
12
// Import types from journal-function crate
13
use journal_function::{
14
- Facets, FileIndexCache, FileIndexCacheBuilder, FileIndexKey, HistogramEngine, Monitor,
15
- Registry, Result as CatalogResult, netdata,
14
+ Facets, FileIndexCache, FileIndexCacheBuilder, FileIndexKey, HistogramEngine, IndexingLimits,
15
+ Monitor, Registry, Result as CatalogResult, netdata,
16
};
17
18
/*
@@ -252,6 +252,7 @@ struct CatalogFunctionInner {
252
cache: FileIndexCache,
253
histogram_engine: Arc<HistogramEngine>,
254
transaction_registry: TransactionRegistry,
255
+ indexing_limits: IndexingLimits,
256
}
257
258
/// Function handler that provides catalog information about journal files
@@ -432,14 +433,13 @@ impl CatalogFunction {
433
/// * `cache_dir` - Directory path for disk cache storage
434
/// * `memory_capacity` - Number of file indexes to keep in memory
435
/// * `disk_capacity` - Disk cache size in bytes
435
- /// * `file_indexing_metrics` - Metrics chart for file indexing operations
436
- /// * `bucket_cache_metrics` - Metrics chart for bucket cache operations
437
- /// * `bucket_operations_metrics` - Metrics chart for bucket operations
436
+ /// * `indexing_limits` - Configuration limits for indexing (cardinality, payload size)
437
pub async fn new(
438
monitor: Monitor,
439
cache_dir: impl Into<std::path::PathBuf>,
440
memory_capacity: usize,
441
disk_capacity: usize,
442
+ indexing_limits: IndexingLimits,
443
) -> CatalogResult<Self> {
444
let registry = Registry::new(monitor);
445
@@ -460,6 +460,7 @@ impl CatalogFunction {
460
cache,
461
histogram_engine: Arc::new(histogram_engine),
462
transaction_registry: TransactionRegistry::new(),
463
+ indexing_limits,
464
};
465
466
Ok(Self {
@@ -532,9 +533,9 @@ impl FunctionHandler for CatalogFunction {
533
})?;
534
let find_files_duration = op_start.elapsed();
535
debug!("[{}] found {} files in time range", txn.id(), files.len(),);
535
- if tracing::enabled!(tracing::Level::DEBUG) {
536
+ if tracing::enabled!(tracing::Level::TRACE) {
537
for (idx, file_info) in files.iter().enumerate() {
537
- debug!(
538
+ tracing::trace!(
539
"[{}] file[{}/{}]: {}",
540
txn.id(),
541
idx + 1,
@@ -556,17 +557,6 @@ impl FunctionHandler for CatalogFunction {
557
facets.len(),
558
facets.precomputed_hash()
559
);
559
- if tracing::enabled!(tracing::Level::DEBUG) {
560
- for (idx, facet) in facets.iter().enumerate() {
561
- debug!(
562
- "[{}] facet[{}/{}]: {}",
563
- txn.id(),
564
- idx + 1,
565
- facets.len(),
566
- facet.as_str(),
567
- );
568
- }
569
- }
560
561
// Build file index keys
562
let source_timestamp_field = FieldName::new_unchecked("_SOURCE_REALTIME_TIMESTAMP");
@@ -583,6 +573,7 @@ impl FunctionHandler for CatalogFunction {
573
keys,
574
&time_range,
575
timeout,
576
+ self.inner.indexing_limits,
577
)
578
.await
579
.map_err(|e| {
@@ -597,9 +588,9 @@ impl FunctionHandler for CatalogFunction {
588
indexed_files.len(),
589
files.len(),
590
);
600
- if tracing::enabled!(tracing::Level::DEBUG) {
591
+ if tracing::enabled!(tracing::Level::TRACE) {
592
for (idx, (key, file_index)) in indexed_files.iter().enumerate() {
602
- debug!(
593
+ tracing::trace!(
594
"[{}] file index[{}/{}]: {}, indexed at: {}, online: {}, bucket duration: {}",
595
txn.id(),
596
idx + 1,
src/crates/netdata-log-viewer/journal-viewer-plugin/src/main.rs
+10
-2
@@ -1,5 +1,6 @@
1
//! journal-viewer-plugin standalone binary
2
3
+use journal_function::IndexingLimits;
4
use journal_registry::Monitor;
5
6
mod catalog;
@@ -43,12 +44,14 @@ async fn run_plugin() -> std::result::Result<(), Box<dyn std::error::Error>> {
44
let config = &plugin_config.config;
45
46
info!(
46
- "configuration loaded: journal_paths={:?}, cache_dir={}, memory_capacity={}, disk_capacity={}, workers={}",
47
+ "configuration loaded: journal_paths={:?}, cache_dir={}, memory_capacity={}, disk_capacity={}, workers={}, max_unique_values_per_field={}, max_field_payload_size={}",
48
config.journal.paths,
49
config.cache.directory,
50
config.cache.memory_capacity,
51
config.cache.disk_capacity,
51
- config.cache.workers
52
+ config.cache.workers,
53
+ config.indexing.max_unique_values_per_field,
54
+ config.indexing.max_field_payload_size
55
);
56
57
let mut runtime = PluginRuntime::new("journal-viewer");
@@ -64,11 +67,16 @@ async fn run_plugin() -> std::result::Result<(), Box<dyn std::error::Error>> {
67
68
// Create catalog function with disk-backed cache
69
info!("creating catalog function with Foyer hybrid cache");
70
+ let indexing_limits = IndexingLimits {
71
+ max_unique_values_per_field: config.indexing.max_unique_values_per_field,
72
+ max_field_payload_size: config.indexing.max_field_payload_size,
73
+ };
74
let catalog_function = CatalogFunction::new(
75
monitor,
76
&config.cache.directory,
77
config.cache.memory_capacity,
78
config.cache.disk_capacity.as_u64() as usize,
79
+ indexing_limits,
80
)
81
.await?;
82
info!("catalog function initialized");
src/crates/netdata-log-viewer/journal-viewer-plugin/src/plugin_config.rs
+53
@@ -64,6 +64,46 @@ impl Default for CacheConfig {
64
}
65
}
66
67
+/// Default value for max_unique_values_per_field
68
+fn default_max_unique_values_per_field() -> usize {
69
+ 500
70
+}
71
+
72
+/// Default value for max_field_payload_size
73
+fn default_max_field_payload_size() -> usize {
74
+ 100
75
+}
76
+
77
+#[derive(Debug, Clone, Serialize, Deserialize)]
78
+#[serde(deny_unknown_fields)]
79
+pub struct IndexingConfig {
80
+ /// Maximum number of unique values to index per field.
81
+ ///
82
+ /// Fields with more unique values than this limit will have their indexing
83
+ /// truncated to prevent unbounded memory growth. This protects against
84
+ /// high-cardinality fields (e.g., MESSAGE with millions of unique values)
85
+ /// causing memory exhaustion during indexing.
86
+ #[serde(default = "default_max_unique_values_per_field")]
87
+ pub max_unique_values_per_field: usize,
88
+
89
+ /// Maximum payload size (in bytes) for field values to index.
90
+ ///
91
+ /// Field values with payloads larger than this limit (or compressed values)
92
+ /// will be skipped. This prevents large binary data or encoded content
93
+ /// from consuming excessive memory.
94
+ #[serde(default = "default_max_field_payload_size")]
95
+ pub max_field_payload_size: usize,
96
+}
97
+
98
+impl Default for IndexingConfig {
99
+ fn default() -> Self {
100
+ Self {
101
+ max_unique_values_per_field: default_max_unique_values_per_field(),
102
+ max_field_payload_size: default_max_field_payload_size(),
103
+ }
104
+ }
105
+}
106
+
107
#[derive(Default, Debug, Clone, Serialize, Deserialize)]
108
#[serde(deny_unknown_fields)]
109
pub struct Config {
@@ -74,6 +114,10 @@ pub struct Config {
114
/// Cache configuration
115
#[serde(rename = "cache")]
116
pub cache: CacheConfig,
117
+
118
+ /// Indexing configuration
119
+ #[serde(rename = "indexing", default)]
120
+ pub indexing: IndexingConfig,
121
}
122
123
pub struct PluginConfig {
@@ -187,6 +231,15 @@ impl PluginConfig {
231
anyhow::bail!("cache.queue_capacity must be greater than 0");
232
}
233
234
+ // Validate indexing configuration
235
+ if config.indexing.max_unique_values_per_field == 0 {
236
+ anyhow::bail!("indexing.max_unique_values_per_field must be greater than 0");
237
+ }
238
+
239
+ if config.indexing.max_field_payload_size == 0 {
240
+ anyhow::bail!("indexing.max_field_payload_size must be greater than 0");
241
+ }
242
+
243
Ok(())
244
}
245
}