[otap-dataflow] fix exclude filter returning empty batch open-telemetry#1483 (open-telemetry#1504)

Fixes open-telemetry#1483

Adds check for fields where an empty vec provided so nothing should be
excluded

Author: c1ly

rust/otap-dataflow/Cargo.toml

@@ -49,7 +49,7 @@ quiver = { package = "otap-df-quiver", path = "crates/quiver" }
 
 ahash = "0.8.11"
 arrayvec = "0.7.6"
-arrow = "57.0"
+arrow = { version = "57.0", features=["prettyprint"] }
 arrow-ipc = { version = "57.0", features=["zstd"] }
 arrow-schema = { version = "57.0" }
 arrow-array = { version = "57.0" }

rust/otap-dataflow/crates/otap/src/filter_processor.rs

@@ -181,7 +181,6 @@ impl local::Processor<OtapPdata> for FilterProcessor {
 mod tests {
     use crate::filter_processor::{FILTER_PROCESSOR_URN, FilterProcessor, config::Config};
     use crate::pdata::OtapPdata;
-    use bytes::BytesMut;
     use otap_df_config::node::NodeUserConfig;
     use otap_df_engine::context::ControllerContext;
     use otap_df_engine::message::Message;
@@ -623,12 +622,11 @@ mod tests {
         move |mut ctx| {
             Box::pin(async move {
                 //convert logsdata to otappdata
-                let mut bytes = BytesMut::new();
+                let mut bytes = vec![];
                 sent.encode(&mut bytes)
                     .expect("failed to encode log data into bytes");
-                let bytes = bytes.freeze();
                 let otlp_logs_bytes =
-                    OtapPdata::new_default(OtlpProtoBytes::ExportLogsRequest(bytes).into());
+                    OtapPdata::new_default(OtlpProtoBytes::ExportLogsRequest(bytes.into()).into());
                 ctx.process(Message::PData(otlp_logs_bytes))
                     .await
                     .expect("failed to process");
@@ -658,13 +656,13 @@ mod tests {
             Box::pin(async move {
                 //convert tracesdata to otappdata
                 let traces_data = build_traces();
-                let mut bytes = BytesMut::new();
+                let mut bytes = vec![];
                 traces_data
                     .encode(&mut bytes)
                     .expect("failed to encode trace data into bytes");
-                let bytes = bytes.freeze();
-                let otlp_traces_bytes =
-                    OtapPdata::new_default(OtlpProtoBytes::ExportTracesRequest(bytes).into());
+                let otlp_traces_bytes = OtapPdata::new_default(
+                    OtlpProtoBytes::ExportTracesRequest(bytes.into()).into(),
+                );
                 ctx.process(Message::PData(otlp_traces_bytes))
                     .await
                     .expect("failed to process");

rust/otap-dataflow/crates/pdata/src/otap/filter/logs.rs

@@ -338,12 +338,30 @@ impl LogMatchProperties {
 
         // invert flag depending on whether we are excluding or including
         if invert {
-            resource_attr_filter =
-                arrow::compute::not(&resource_attr_filter).expect("not doesn't fail");
+            // default filter is all true
 
-            log_record_filter = arrow::compute::not(&log_record_filter).expect("not doesn't fail");
+            // if no resource_attributes to filter on are defined then we can ignore them
+            // that is we will resort to the default filter otherwise we can invert if the flag is set
+            if !self.resource_attributes.is_empty() {
+                resource_attr_filter =
+                    arrow::compute::not(&resource_attr_filter).expect("not doesn't fail");
+            }
+
+            // if no log records fields to filter on are defined then we can ignore them
+            // that is we will resort to the default filter otherwise we can invert if the flag is set
+            if !self.bodies.is_empty()
+                || !self.severity_texts.is_empty()
+                || self.severity_number.is_some()
+            {
+                log_record_filter =
+                    arrow::compute::not(&log_record_filter).expect("not doesn't fail");
+            }
 
-            log_attr_filter = arrow::compute::not(&log_attr_filter).expect("not doesn't fail");
+            // if no record_attributes to filter on are defined then we can ignore them
+            // that is we will resort to the default filter otherwise we can invert if the flag is set
+            if !self.record_attributes.is_empty() {
+                log_attr_filter = arrow::compute::not(&log_attr_filter).expect("not doesn't fail");
+            }
         }
 
         Ok((resource_attr_filter, log_record_filter, log_attr_filter))
@@ -583,4 +601,54 @@ mod test {
 
         assert_equivalent(&[otap_to_otlp(&result)], &[otap_to_otlp(&expected)]);
     }
+
+    #[test]
+    fn test_filter_exclude_no_attributes() {
+        // Filter only for WARN logs
+        let exclude = LogMatchProperties::new(
+            MatchType::Strict,
+            Vec::new(),
+            Vec::new(),
+            vec!["WARN".into()],
+            None,
+            Vec::new(),
+        );
+
+        let filter = LogFilter::new(None, Some(exclude), Vec::new());
+
+        let log_records = vec![
+            LogRecord::build().severity_text("WARN").finish(),
+            LogRecord::build().severity_text("WARN").finish(),
+            LogRecord::build().severity_text("INFO").finish(),
+            LogRecord::build().severity_text("INFO").finish(),
+        ];
+
+        let logs_data = LogsData {
+            resource_logs: vec![ResourceLogs {
+                scope_logs: vec![ScopeLogs {
+                    log_records: log_records.clone(),
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }],
+        };
+
+        let input = otlp_to_otap(&OtlpProtoMessage::Logs(logs_data));
+
+        let (result, logs_consumed, logs_filtered) = filter.filter(input).unwrap();
+        assert_eq!(logs_consumed, 4);
+        assert_eq!(logs_filtered, 2);
+
+        let expected = otlp_to_otap(&OtlpProtoMessage::Logs(LogsData {
+            resource_logs: vec![ResourceLogs {
+                scope_logs: vec![ScopeLogs {
+                    log_records: log_records[2..4].to_vec(),
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }],
+        }));
+
+        assert_equivalent(&[otap_to_otlp(&result)], &[otap_to_otlp(&expected)]);
+    }
 }

rust/otap-dataflow/crates/pdata/src/otap/filter/traces.rs

@@ -562,20 +562,48 @@ impl TraceMatchProperties {
 
         // invert flag depending on whether we are excluding or including
         if invert {
-            resource_attr_filter =
-                arrow::compute::not(&resource_attr_filter).expect("not doesn't fail");
+            // default filter is all true
 
-            span_filter = arrow::compute::not(&span_filter).expect("not doesn't fail");
+            // if no resource_attributes to filter on are defined then we can ignore them
+            // that is we will resort to the default filter otherwise we can invert if the flag is set
+            if !self.resource_attributes.is_empty() {
+                resource_attr_filter =
+                    arrow::compute::not(&resource_attr_filter).expect("not doesn't fail");
+            }
 
-            span_attr_filter = arrow::compute::not(&span_attr_filter).expect("not doesn't fail");
+            // if no span_names to filter on are defined then we can ignore them
+            // that is we will resort to the default filter otherwise we can invert if the flag is set
+            if !self.span_names.is_empty() {
+                span_filter = arrow::compute::not(&span_filter).expect("not doesn't fail");
+            }
 
-            span_event_filter = arrow::compute::not(&span_event_filter).expect("not doesn't fail");
+            // if no span_attributes to filter on are defined then we can ignore them
+            // that is we will resort to the default filter otherwise we can invert if the flag is set
+            if !self.span_attributes.is_empty() {
+                span_attr_filter =
+                    arrow::compute::not(&span_attr_filter).expect("not doesn't fail");
+            }
 
-            span_event_attr_filter =
-                arrow::compute::not(&span_event_attr_filter).expect("not doesn't fail");
+            // if no event_names to filter on are defined then we can ignore them
+            // that is we will resort to the default filter otherwise we can invert if the flag is set
+            if !self.event_names.is_empty() {
+                span_event_filter =
+                    arrow::compute::not(&span_event_filter).expect("not doesn't fail");
+            }
 
-            span_link_attr_filter =
-                arrow::compute::not(&span_link_attr_filter).expect("not doesn't fail");
+            // if no event_attributes to filter on are defined then we can ignore them
+            // that is we will resort to the default filter otherwise we can invert if the flag is set
+            if !self.event_attributes.is_empty() {
+                span_event_attr_filter =
+                    arrow::compute::not(&span_event_attr_filter).expect("not doesn't fail");
+            }
+
+            // if no link_attributes to filter on are defined then we can ignore them
+            // that is we will resort to the default filter otherwise we can invert if the flag is set
+            if !self.link_attributes.is_empty() {
+                span_link_attr_filter =
+                    arrow::compute::not(&span_link_attr_filter).expect("not doesn't fail");
+            }
         }
 
         Ok((
@@ -689,7 +717,7 @@ mod test {
     use crate::testing::round_trip::{otap_to_otlp, otlp_to_otap};
 
     #[test]
-    fn test_filter_no_attributes() {
+    fn test_filter_include_no_attributes() {
         let include = TraceMatchProperties::new(
             MatchType::Strict,
             Vec::new(),
@@ -737,4 +765,54 @@ mod test {
 
         assert_equivalent(&[otap_to_otlp(&result)], &[otap_to_otlp(&expected)]);
     }
+
+    #[test]
+    fn test_filter_exclude_no_attributes() {
+        let exclude = TraceMatchProperties::new(
+            MatchType::Strict,
+            Vec::new(),
+            Vec::new(),
+            vec!["span_name_1".into()],
+            Vec::new(),
+            Vec::new(),
+            Vec::new(),
+        );
+
+        let filter = TraceFilter::new(None, Some(exclude));
+
+        let spans = vec![
+            Span::build().name("span_name_1").finish(),
+            Span::build().name("span_name_1").finish(),
+            Span::build().name("span_name_2").finish(),
+            Span::build().name("span_name_2").finish(),
+        ];
+
+        let traces_data = TracesData {
+            resource_spans: vec![ResourceSpans {
+                scope_spans: vec![ScopeSpans {
+                    spans: spans.clone(),
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }],
+        };
+
+        let input = otlp_to_otap(&OtlpProtoMessage::Traces(traces_data));
+        let (result, spans_consumed, spans_filtered) = filter.filter(input).unwrap();
+
+        assert_eq!(spans_consumed, 4);
+        assert_eq!(spans_filtered, 2);
+
+        let expected = otlp_to_otap(&OtlpProtoMessage::Traces(TracesData {
+            resource_spans: vec![ResourceSpans {
+                scope_spans: vec![ScopeSpans {
+                    spans: spans[2..4].to_vec(),
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }],
+        }));
+
+        assert_equivalent(&[otap_to_otlp(&result)], &[otap_to_otlp(&expected)]);
+    }
 }