drain_flow/record/
mod.rs

1// Copyright Nicholas Harring. All rights reserved.
2//
3// This program is free software: you can redistribute it and/or modify it under
4// the terms of the Server Side Public License, version 1, as published by MongoDB, Inc.
5// This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
6// without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
7// See the Server Side Public License for more details. You should have received a copy of the
8// Server Side Public License along with this program.
9// If not, see <http://www.mongodb.com/licensing/server-side-public-license>.
10
11pub mod tokens;
12extern crate derive_more;
13
14use std::fmt;
15
16use lazy_static::lazy_static;
17use string_interner::DefaultSymbol;
18use tracing::{debug, instrument};
19use uuid::Uuid;
20
21use self::tokens::{Offset, Token, TokenStream}; // Added Offset, removed TypedToken
22use crate::drains::simple::INTERNER;
23
24lazy_static! {
25    pub static ref ASTERISK: DefaultSymbol = INTERNER.write().get_or_intern_static("<*>");
26}
27/// Represents a processed log record.
28///
29/// A `Record` consists of a `TokenStream` (the tokenized log line) and a
30/// unique identifier (`Uuid`).
31#[derive(Clone, Debug)]
32pub struct Record {
33    pub(crate) inner: TokenStream,
34    pub uid: Uuid,
35}
36impl Record {
37    /// Creates a new `Record` from a given log line string.
38    ///
39    /// This involves tokenizing the line and generating a new UUID (version 1)
40    /// for the record.
41    ///
42    /// # Arguments
43    ///
44    /// * `line` - The raw log line as a `String`.
45    ///
46    /// # Returns
47    ///
48    /// A new `Record` instance.
49    #[instrument(name = "Create new record", level = "trace", skip(line))]
50    pub fn new(line: String) -> Self {
51        // Generate a V1 UUID (timestamp-based)
52        // Requires a timestamp and a 16-byte node ID.
53        // For the node ID, we can use a constant byte array.
54        // The uniqueness of the node ID is not critical for this application.
55        let now = chrono::Utc::now();
56        let context = uuid::NoContext; // Added context for clock sequence
57        let timestamp = uuid::v1::Timestamp::from_unix(
58            context, // Added context as the first argument
59            now.timestamp() as u64,
60            now.timestamp_subsec_nanos(),
61        );
62        // Example node ID, can be any 6 bytes.
63        const NODE_ID: &[u8; 6] = b"drainf";
64        Self {
65            inner: TokenStream::from_unicode_line(&line),
66            uid: Uuid::new_v1(timestamp, NODE_ID),
67        }
68    }
69
70    /// Calculates a similarity score between this `Record` (acting as a template)
71    /// and a `candidate` `Record`.
72    ///
73    /// The score is the number of matching tokens between the two records.
74    /// A `Token::Wildcard` in the template matches any token in the candidate.
75    ///
76    /// # Arguments
77    ///
78    /// * `candidate` - The `Record` to compare against this record.
79    ///
80    /// # Returns
81    ///
82    /// The similarity score as a `u64`.
83    #[instrument(
84        name = "Calculate similarity score",
85        level = "trace",
86        skip(candidate, self)
87    )]
88    pub fn calc_sim_score(&self, candidate: &Record) -> u64 {
89        // self is the log group's event record (template), candidate is the new log line.
90        // The iterator for `self` (template) should yield Tokens.
91        // The iterator for `candidate` (new line) can yield resolved Strings or Tokens.
92        // For simplicity, let's assume both yield Tokens for comparison.
93        self.inner
94            .inner
95            .iter() // Iterate over (Offset, Token) pairs in the template
96            .zip(candidate.inner.inner.iter()) // Iterate over (Offset, Token) pairs in the candidate
97            .filter(|((_, template_token), (_, candidate_token))| {
98                match template_token {
99                    Token::Wildcard => {
100                        debug!(
101                            "template token is Wildcard, matches candidate token {:?}",
102                            candidate_token
103                        );
104                        true // Wildcard in template matches any token in candidate
105                    }
106                    _ => {
107                        // For non-wildcard tokens, they must be equal.
108                        // This comparison depends on how PartialEq is implemented for Token.
109                        // Assuming Token::Value(TypedToken::String(Symbol)) comparison works.
110                        if template_token == candidate_token {
111                            debug!(
112                                "template token {:?} matches candidate token {:?}",
113                                template_token, candidate_token
114                            );
115                            true
116                        } else {
117                            debug!(
118                                "template token {:?} does NOT match candidate token {:?}",
119                                template_token, candidate_token
120                            );
121                            false
122                        }
123                    }
124                }
125            })
126            .count() as u64 // Count the number of matching token pairs
127    }
128
129    /// Returns the first token of the record's `TokenStream`.
130    ///
131    /// # Returns
132    ///
133    /// An `Option<DefaultSymbol>` containing the first token if it exists, otherwise `None`.
134    #[instrument(level = "trace", skip(self))]
135    pub fn first(&self) -> Option<DefaultSymbol> {
136        self.inner.first().map(std::convert::Into::into)
137    }
138
139    /// Returns the number of tokens in the record's `TokenStream`.
140    ///
141    /// # Returns
142    ///
143    /// The length of the token stream as a `usize`.
144    #[instrument(level = "trace", skip(self))]
145    pub fn len(&self) -> usize {
146        self.inner.len()
147    }
148
149    /// Checks if the record's `TokenStream` is empty.
150    ///
151    /// # Returns
152    ///
153    /// `true` if the token stream is empty, `false` otherwise.
154    #[instrument(level = "trace", skip(self))]
155    pub fn is_empty(&self) -> bool {
156        self.inner.len() == 0
157    }
158
159    /// Resolves a `DefaultSymbol` back to its original string representation.
160    ///
161    /// This is a utility method that uses the global string interner to retrieve
162    /// the string associated with a given symbol.
163    ///
164    /// # Arguments
165    ///
166    /// * `sym` - The `DefaultSymbol` to resolve.
167    ///
168    /// # Returns
169    ///
170    /// An `Option<String>` containing the resolved string if the symbol exists,
171    /// otherwise `None`.
172    #[instrument(level = "trace")]
173    pub fn resolve(sym: DefaultSymbol) -> Option<String> {
174        INTERNER
175            .read()
176            .resolve(sym)
177            .map(std::borrow::ToOwned::to_owned)
178    }
179}
180
181/// An iterator that consumes a `Record` and yields its tokens as `String`s.
182pub struct IntoIter {
183    record: Record,
184    index: usize,
185}
186
187impl Iterator for IntoIter {
188    type Item = String;
189
190    fn next(&mut self) -> Option<String> {
191        if self.index >= self.record.len() {
192            return None;
193        }
194        // Use .to_string() which now correctly handles <*> for Token::Wildcard
195        let token_display = self
196            .record
197            .inner
198            .get_token_at_index(self.index)
199            .map(|t| t.to_string());
200
201        self.index += 1;
202        token_display
203    }
204}
205
206impl IntoIterator for Record {
207    type IntoIter = IntoIter;
208    type Item = String; // This iterator yields Strings, used by old calc_sim_score
209
210    fn into_iter(self) -> Self::IntoIter {
211        IntoIter {
212            record: self, // Consumes the record
213            index: 0,
214        }
215    }
216}
217
218impl<'a> IntoIterator for &'a Record {
219    type Item = &'a Token; // Yields references to Tokens
220    type IntoIter =
221        std::iter::Map<std::slice::Iter<'a, (Offset, Token)>, fn(&(Offset, Token)) -> &Token>;
222
223    fn into_iter(self) -> Self::IntoIter {
224        self.inner.inner.iter().map(|(_, token)| token)
225    }
226}
227
228impl fmt::Display for Record {
229    /// Formats the `Record` for display.
230    ///
231    /// This implementation reconstructs the original log line from the `TokenStream`,
232    /// preserving whitespace.
233    ///
234    /// # Arguments
235    ///
236    /// * `f` - The formatter to write into.
237    ///
238    /// # Returns
239    ///
240    /// A `fmt::Result` indicating success or failure of the formatting operation.
241    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
242        write!(f, "{}", self.inner)
243    }
244}
245#[cfg(test)]
246mod should {
247    use joinery::{Joinable, JoinableIterator};
248    use proptest::{prelude::*, string::string_regex};
249    use spectral::prelude::*;
250
251    use crate::{drains::simple::INTERNER, record::Record};
252
253    prop_compose! {
254        fn gen_word()(s in "[[:alpha:]]+") -> String {
255            s
256        }
257    }
258
259    fn gen_variable_string() -> impl Strategy<Value = String> {
260        prop_oneof![
261            // UUID
262            string_regex(r"[A-Fa-f0-9]{8}-(?:[A-Fa-f0-9]{4}-){3}[A-Fa-f0-9]{12}").unwrap(),
263            // MAC address
264            string_regex(r"(?:(?:[A-Fa-f0-9]{2}:){5}[A-Fa-f0-9]{2})").unwrap(),
265            // IPv6
266            string_regex(r"((([0-9A-Fa-f]{1,4}:){7}([0-9A-Fa-f]{1,4}|:))|(([0-9A-Fa-f]{1,4}:){6}(:[0-9A-Fa-f]{1,4}|((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})|:))|(([0-9A-Fa-f]{1,4}:){5}(((:[0-9A-Fa-f]{1,4}){1,2})|:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})|:))|(([0-9A-Fa-f]{1,4}:){4}(((:[0-9A-Fa-f]{1,4}){1,3})|((:[0-9A-Fa-f]{1,4})?:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){3}(((:[0-9A-Fa-f]{1,4}){1,4})|((:[0-9A-Fa-f]{1,4}){0,2}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){2}(((:[0-9A-Fa-f]{1,4}){1,5})|((:[0-9A-Fa-f]{1,4}){0,3}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){1}(((:[0-9A-Fa-f]{1,4}){1,6})|((:[0-9A-Fa-f]{1,4}){0,4}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(:(((:[0-9A-Fa-f]{1,4}){1,7})|((:[0-9A-Fa-f]{1,4}){0,5}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:)))").unwrap(),
267            // Base 10 Integer
268            string_regex(r"(?:[+-]?(?:[0-9]+))").unwrap(),
269        ]
270    }
271
272    fn gen_phrase(len: usize) -> impl Strategy<Value = String> {
273        prop::collection::vec(gen_word(), len)
274            .prop_flat_map(|vec| Just(vec.iter().join_with(" ").to_string()))
275    }
276
277    fn gen_vars(len: usize) -> impl Strategy<Value = String> {
278        prop::collection::vec(gen_variable_string(), len)
279            .prop_flat_map(|vec| Just(vec.iter().join_with(" ").to_string()))
280    }
281
282    fn gen_complex(base: usize, variable: usize) -> impl Strategy<Value = String> {
283        let base = gen_phrase(base);
284        let vars = gen_vars(variable);
285        (base, vars).prop_map(|(b, v)| [b, v].join_with(" ").to_string())
286    }
287
288    prop_compose! {
289        fn gen_matching_lines(base_len: usize, var_count: usize, num_lines: usize)(base_phrase in gen_phrase(base_len), var_set in prop::collection::vec(gen_vars(var_count), num_lines)) -> Vec<String> {
290            var_set.iter().map(|v| {[base_phrase.clone(), v.to_string()].join_with(" ").to_string()}).collect::<Vec<String>>()
291        }
292    }
293
294    proptest! {
295        #[test]
296        fn test_proptest_base_record_new(phrase in gen_phrase(5)) {
297            let rec = Record::new(phrase.clone());
298            prop_assert_eq!(phrase, rec.to_string());
299        }
300    }
301
302    proptest! {
303        #[test]
304        fn test_proptest_variable_record_new(line in gen_complex(7, 3)) {
305            // Because we don't try to fully preserve whitespace semantics
306            // instead we test that the stringified form of the record is "stable"
307            let rec = Record::new(line.clone());
308            let rec2 = Record::new(rec.to_string());
309            prop_assert_eq!(rec.to_string(), rec2.to_string());
310
311            // Whitespace internally is preserved, only the end is missing
312            let reconstituted = rec.to_string();
313            prop_assert!(line.contains(&reconstituted));
314        }
315    }
316
317    proptest! {
318        #[test]
319        fn test_matching_records(lines in gen_matching_lines(7, 3, 3)) {
320            let recs = lines
321                .iter()
322                .map(|l| Record::new(l.clone()))
323                .collect::<Vec<Record>>();
324            let base = recs[0].clone();
325            let score1 = base.calc_sim_score(&recs[1].clone());
326            let score2 = base.calc_sim_score(&recs[2].clone());
327
328            // Each generated line shares a seven-word prefix with the base line.
329            // Additional tokens may coincidentally match, so we only assert the
330            // similarity is at least that prefix length.
331            prop_assert!(score1 >= 7 && score2 >= 7);
332        }
333    }
334
335    #[test]
336    fn test_record_first() {
337        let input = "Message send failed to remote host: foo.bar.com".to_string();
338        let rec = Record::new(input);
339        let val = rec.first().unwrap();
340        assert_eq!(INTERNER.read().resolve(val).unwrap(), "Message");
341    }
342
343    #[test]
344    fn test_record_len() {
345        let input = "Message send failed to remote host: foo.bar.com".to_string();
346        let rec = Record::new(input);
347        assert_eq!(rec.len(), 7);
348    }
349
350    #[test]
351    fn test_consuming_iter() {
352        let input = "Message send failed to remote host: foo.bar.com".to_string();
353        let rec = Record::new(input.clone());
354        let tokens = rec.into_iter().collect::<Vec<String>>();
355        let words = &input
356            .split(|c: char| c.is_whitespace())
357            .map(|s| s.to_owned())
358            .collect::<Vec<String>>();
359        assert_that(&tokens.iter()).contains_all_of(&words.iter());
360    }
361
362    #[test]
363    fn test_non_consuming_iter() {
364        let input = "Message send failed to remote host: foo.bar.com".to_string();
365        let rec = Record::new(input);
366        let tokens = (&rec).into_iter().collect::<Vec<_>>();
367        // Disambiguate has_length by using .len() and asserting equality
368        assert_that(&tokens.len()).is_equal_to(7);
369    }
370}
drain_flow/record/mod.rs

drain_flow/record/
mod.rs
