1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
//! Region processor — evaluates log records against region definitions.
#[cfg(test)]
#[path = "processor_tests.rs"]
mod processor_tests;
use super::config::{CompiledMatchPoint, RegionDefinition};
use super::Region;
use crate::filter::eval;
use crate::record::LogRecord;
use std::collections::HashMap;
use tracing::instrument;
/// A pending start point awaiting a matching end.
#[derive(Debug, Clone)]
struct PendingStart {
/// Index of the start record in the LogStore.
record_index: usize,
/// Timestamp of the start record.
timestamp: chrono::DateTime<chrono::Utc>,
/// Extracted metadata from the start record.
metadata: HashMap<String, String>,
/// Rendered reason from the matched start point.
reason: Option<String>,
}
/// Processes log records against region definitions to detect regions.
pub struct RegionProcessor {
definitions: Vec<RegionDefinition>,
/// Pending starts per definition index.
pending: Vec<Vec<PendingStart>>,
/// Detected regions.
regions: Vec<Region>,
/// Next record index to process (for incremental processing).
next_index: usize,
}
impl RegionProcessor {
/// Create a new processor with the given region definitions.
pub fn new(definitions: Vec<RegionDefinition>) -> Self {
let pending = vec![Vec::new(); definitions.len()];
Self {
definitions,
pending,
regions: Vec::new(),
next_index: 0,
}
}
/// Process a batch of records from the store.
/// Records are processed in order; call repeatedly for incremental processing.
/// Returns the records that were tagged with region metadata (indices).
#[instrument(skip(self, records), fields(record_count = records.len()))]
pub fn process_records(&mut self, records: &[LogRecord]) {
for (i, record) in records.iter().enumerate() {
let absolute_index = self.next_index + i;
for def_idx in 0..self.definitions.len() {
// Check END points first
if let Some((end_meta, end_reason)) =
try_match_points(&self.definitions[def_idx].end_points, record)
{
// Try correlation with pending starts
if let Some(region) =
self.try_correlate(def_idx, absolute_index, end_meta, end_reason, record)
{
self.regions.push(region);
}
}
// Check START points
if let Some((start_meta, start_reason)) =
try_match_points(&self.definitions[def_idx].start_points, record)
{
// Create timed-out regions for expired pending starts
if let Some(timeout) = self.definitions[def_idx].timeout {
let ts = records[i].timestamp;
let timeout_chrono =
chrono::Duration::from_std(timeout).unwrap_or(chrono::Duration::MAX);
let (expired, remaining): (Vec<_>, Vec<_>) = self.pending[def_idx]
.drain(..)
.partition(|p| ts.signed_duration_since(p.timestamp) >= timeout_chrono);
self.pending[def_idx] = remaining;
// Create timed-out regions for expired starts
let def = &self.definitions[def_idx];
for pending in expired {
let timeout_reason = def
.timeout_reason
.as_ref()
.map(|t| super::config::render_template(t, &pending.metadata));
let mut metadata = pending.metadata.clone();
if let Some(tr) = &timeout_reason {
metadata.insert("end_reason".to_string(), tr.clone());
}
if let Some(sr) = &pending.reason {
let rendered =
super::config::render_template(sr, &pending.metadata);
metadata.insert("start_reason".to_string(), rendered.clone());
}
let name =
super::config::render_template(&def.name_template, &metadata);
let description = def
.description_template
.as_ref()
.map(|t| super::config::render_template(t, &metadata));
self.regions.push(Region {
definition_name: def.name.clone(),
name,
description,
start_reason: pending
.reason
.as_ref()
.map(|r| super::config::render_template(r, &pending.metadata)),
end_reason: timeout_reason,
start_index: pending.record_index,
end_index: pending.record_index, // timed-out: end = start
metadata,
timed_out: true,
});
}
}
self.pending[def_idx].push(PendingStart {
record_index: absolute_index,
timestamp: records[i].timestamp,
metadata: start_meta,
reason: start_reason,
});
}
}
}
self.next_index += records.len();
}
/// Try to correlate an end match with a pending start.
fn try_correlate(
&mut self,
def_idx: usize,
end_index: usize,
end_meta: HashMap<String, String>,
end_reason: Option<String>,
_end_record: &LogRecord,
) -> Option<Region> {
let def = &self.definitions[def_idx];
let pendings = &mut self.pending[def_idx];
// Walk backwards (LIFO) through pending starts
let mut found_idx = None;
for (i, pending) in pendings.iter().enumerate().rev() {
// Check timeout
if let Some(timeout) = def.timeout {
let elapsed = _end_record
.timestamp
.signed_duration_since(pending.timestamp);
if elapsed >= chrono::Duration::from_std(timeout).unwrap_or(chrono::Duration::MAX) {
continue;
}
}
if def.correlate.is_empty() {
// No correlation fields → nearest pending (LIFO)
found_idx = Some(i);
break;
}
// Check all correlate fields match
let all_match = def.correlate.iter().all(|field| {
let start_val = pending.metadata.get(field);
let end_val = end_meta.get(field);
match (start_val, end_val) {
(Some(s), Some(e)) => s == e,
_ => false,
}
});
if all_match {
found_idx = Some(i);
break;
}
}
let found_idx = found_idx?;
let pending = pendings.remove(found_idx);
// Merge metadata (start + end, end overwrites on conflict)
let mut metadata = pending.metadata.clone();
metadata.extend(end_meta);
// Add start_reason and end_reason to metadata for template rendering
let start_reason = pending
.reason
.as_ref()
.map(|r| super::config::render_template(r, &pending.metadata));
let end_reason_rendered = end_reason
.as_ref()
.map(|r| super::config::render_template(r, &metadata));
if let Some(sr) = &start_reason {
metadata.insert("start_reason".to_string(), sr.clone());
}
if let Some(er) = &end_reason_rendered {
metadata.insert("end_reason".to_string(), er.clone());
}
let name = super::config::render_template(&def.name_template, &metadata);
let description = def
.description_template
.as_ref()
.map(|t| super::config::render_template(t, &metadata));
Some(Region {
definition_name: def.name.clone(),
name,
description,
start_reason,
end_reason: end_reason_rendered,
start_index: pending.record_index,
end_index,
metadata,
timed_out: false,
})
}
/// Get all detected regions.
pub fn regions(&self) -> &[Region] {
&self.regions
}
/// Get the number of detected regions.
pub fn region_count(&self) -> usize {
self.regions.len()
}
/// Get pending start count (for diagnostics).
pub fn pending_count(&self) -> usize {
self.pending.iter().map(|p| p.len()).sum()
}
}
/// Try to match a record against a list of match points.
/// Returns extracted metadata and rendered reason if any point matches.
fn try_match_points(
points: &[CompiledMatchPoint],
record: &LogRecord,
) -> Option<(HashMap<String, String>, Option<String>)> {
for point in points {
if eval::eval(&point.filter, record) {
let mut metadata = HashMap::new();
// Extract metadata via regex
if let Some(re) = &point.regex {
if let Some(caps) = re.captures(&record.message) {
for name in re.capture_names().flatten() {
if let Some(m) = caps.name(name) {
metadata.insert(name.to_string(), m.as_str().to_string());
}
}
}
}
return Some((metadata, point.reason.clone()));
}
}
None
}
// tag_record removed: region metadata now lives in RegionStore, not on LogRecord.