Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix(reduce transform): improve array handling #3076

Merged
merged 3 commits into from
Jul 16, 2020
Merged

fix(reduce transform): improve array handling #3076

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f8f4e50
fix(reduce transform): improve array handling
lukesteensen Jul 15, 2020
e2c7740
fix clippy
lukesteensen Jul 15, 2020
f34ed98
concat arrays
lukesteensen Jul 15, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
52 changes: 44 additions & 8 deletions src/transforms/reduce/merge_strategy.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,16 +74,45 @@ impl ReduceValueMerger for ConcatMerger {
//------------------------------------------------------------------------------

#[derive(Debug, Clone)]
struct ArrayMerger {
struct ConcatArrayMerger {
v: Vec<Value>,
}

impl ArrayMerger {
impl ConcatArrayMerger {
fn new(v: Vec<Value>) -> Self {
Self { v }
}
}

impl ReduceValueMerger for ConcatArrayMerger {
fn add(&mut self, v: Value) -> Result<(), String> {
if let Value::Array(a) = v {
self.v.extend_from_slice(&a);
} else {
self.v.push(v);
}
Ok(())
}

fn insert_into(self: Box<Self>, k: String, v: &mut LogEvent) -> Result<(), String> {
v.insert(k, Value::Array(self.v));
Ok(())
}
}

//------------------------------------------------------------------------------

#[derive(Debug, Clone)]
struct ArrayMerger {
v: Vec<Value>,
}

impl ArrayMerger {
fn new(v: Value) -> Self {
Self { v: vec![v] }
}
}

impl ReduceValueMerger for ArrayMerger {
fn add(&mut self, v: Value) -> Result<(), String> {
self.v.push(v);
Expand Down Expand Up @@ -370,15 +399,13 @@ pub fn get_value_merger(v: Value, m: &MergeStrategy) -> Result<Box<dyn ReduceVal
},
MergeStrategy::Concat => match v {
Value::Bytes(b) => Ok(Box::new(ConcatMerger::new(b))),
Value::Array(a) => Ok(Box::new(ConcatArrayMerger::new(a))),
_ => Err(format!(
"expected string value, found: '{}'",
"expected string or array value, found: '{}'",
v.to_string_lossy()
)),
},
MergeStrategy::Array => match v {
Value::Array(a) => Ok(Box::new(ArrayMerger::new(a))),
_ => Ok(Box::new(ArrayMerger::new(vec![v]))),
},
MergeStrategy::Array => Ok(Box::new(ArrayMerger::new(v))),
MergeStrategy::Discard => Ok(Box::new(DiscardMerger::new(v))),
}
}
Expand Down Expand Up @@ -432,7 +459,7 @@ mod test {
assert!(get_value_merger(json!([]).into(), &MergeStrategy::Max).is_err());
assert!(get_value_merger(json!([]).into(), &MergeStrategy::Min).is_err());
assert!(get_value_merger(json!([]).into(), &MergeStrategy::Array).is_ok());
assert!(get_value_merger(json!([]).into(), &MergeStrategy::Concat).is_err());
assert!(get_value_merger(json!([]).into(), &MergeStrategy::Concat).is_ok());

assert!(get_value_merger(json!({}).into(), &MergeStrategy::Discard).is_ok());
assert!(get_value_merger(json!({}).into(), &MergeStrategy::Sum).is_err());
Expand Down Expand Up @@ -512,6 +539,15 @@ mod test {
merge(4.3.into(), 4.2.into(), &MergeStrategy::Min),
Ok(4.2.into())
);

assert_eq!(
merge(json!([4]).into(), json!([2]).into(), &MergeStrategy::Concat),
Ok(json!([4, 2]).into())
);
assert_eq!(
merge(json!([]).into(), 42.into(), &MergeStrategy::Concat),
Ok(json!([42]).into())
);
}

fn merge(initial: Value, additional: Value, strategy: &MergeStrategy) -> Result<Value, String> {
Expand Down
94 changes: 88 additions & 6 deletions src/transforms/reduce/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,33 +77,32 @@ impl ReduceState {
fn new(e: LogEvent, strategies: &IndexMap<String, MergeStrategy>) -> Self {
Self {
stale_since: Instant::now(),
// TODO: all_fields alternative that consumes
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As it was foretold.

fields: e
.all_fields()
.into_iter()
.filter_map(|(k, v)| {
if let Some(strat) = strategies.get(&k) {
match get_value_merger(v.clone(), strat) {
match get_value_merger(v, strat) {
Ok(m) => Some((k, m)),
Err(err) => {
warn!("failed to create merger for field '{}': {}", k, err);
None
}
}
} else {
Some((k, v.clone().into()))
Some((k, v.into()))
}
})
.collect(),
}
}

fn add_event(&mut self, e: LogEvent, strategies: &IndexMap<String, MergeStrategy>) {
for (k, v) in e.all_fields() {
for (k, v) in e.into_iter() {
let strategy = strategies.get(&k);
match self.fields.entry(k) {
hash_map::Entry::Vacant(entry) => {
if let Some(strat) = strategy {
match get_value_merger(v.clone(), strat) {
match get_value_merger(v, strat) {
Ok(m) => {
entry.insert(m);
}
Expand Down Expand Up @@ -291,6 +290,7 @@ mod test {
topology::config::{TransformConfig, TransformContext},
Event,
};
use serde_json::json;

#[test]
fn reduce_from_condition() {
Expand Down Expand Up @@ -493,4 +493,86 @@ identifier_fields = [ "request_id" ]
Value::from(7)
);
}

#[test]
fn arrays() {
let mut reduce = toml::from_str::<ReduceConfig>(
r#"
identifier_fields = [ "request_id" ]

merge_strategies.foo = "array"
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about one for concat?

merge_strategies.bar = "concat"

[ends_when]
"test_end.exists" = true
"#,
)
.unwrap()
.build(TransformContext::new_test())
.unwrap();

let mut outputs = Vec::new();

let mut e = Event::from("test message 1");
e.as_mut_log().insert("foo", json!([1, 3]));
e.as_mut_log().insert("bar", json!([1, 3]));
e.as_mut_log().insert("request_id", "1");
reduce.transform_into(&mut outputs, e);

let mut e = Event::from("test message 2");
e.as_mut_log().insert("foo", json!([2, 4]));
e.as_mut_log().insert("bar", json!([2, 4]));
e.as_mut_log().insert("request_id", "2");
reduce.transform_into(&mut outputs, e);

let mut e = Event::from("test message 3");
e.as_mut_log().insert("foo", json!([5, 7]));
e.as_mut_log().insert("bar", json!([5, 7]));
e.as_mut_log().insert("request_id", "1");
reduce.transform_into(&mut outputs, e);

let mut e = Event::from("test message 4");
e.as_mut_log().insert("foo", json!("done"));
e.as_mut_log().insert("bar", json!("done"));
e.as_mut_log().insert("request_id", "1");
e.as_mut_log().insert("test_end", "yep");
reduce.transform_into(&mut outputs, e);

assert_eq!(outputs.len(), 1);
assert_eq!(
outputs.first().unwrap().as_log()[&"foo".into()],
json!([[1, 3], [5, 7], "done"]).into()
);

assert_eq!(outputs.len(), 1);
assert_eq!(
outputs.first().unwrap().as_log()[&"bar".into()],
json!([1, 3, 5, 7, "done"]).into()
);

outputs.clear();

let mut e = Event::from("test message 5");
e.as_mut_log().insert("foo", json!([6, 8]));
e.as_mut_log().insert("bar", json!([6, 8]));
e.as_mut_log().insert("request_id", "2");
reduce.transform_into(&mut outputs, e);

let mut e = Event::from("test message 6");
e.as_mut_log().insert("foo", json!("done"));
e.as_mut_log().insert("bar", json!("done"));
e.as_mut_log().insert("request_id", "2");
e.as_mut_log().insert("test_end", "yep");
reduce.transform_into(&mut outputs, e);

assert_eq!(outputs.len(), 1);
assert_eq!(
outputs.first().unwrap().as_log()[&"foo".into()],
json!([[2, 4], [6, 8], "done"]).into()
);
assert_eq!(
outputs.first().unwrap().as_log()[&"bar".into()],
json!([2, 4, 6, 8, "done"]).into()
);
}
}