[otap-dataflow]basic filter processor logs followup (#1341)

Add a basic filter processor that is based on the golang version minus
the OTTL support

Update to #1279 
- fixed edge case where optional record batches are not present
- fixed edge case where logs with no resource attribute/record
attributes are not being filtered out
- fixed string column not guaranteed to be a string array when downcast
to apply regex filter
- utilized booleanbuilder to optimize building the id filter

Add support for log filtering

Defined to following rust structs that will allow us to filter on 
- resource attributes
- log record attributes
- log record bodies
- log record severity text
- log record severity number (optional)

```rust
/// struct that describes the overall requirements to use in order to filter logs
#[derive(Debug, Clone, Deserialize)]
pub struct LogFilter {
    // Include match properties describe logs that should be included in the Collector Service pipeline,
    // all other logs should be dropped from further processing.
    // If both Include and Exclude are specified, Include filtering occurs first.
    include: LogMatchProperties,
    // Exclude match properties describe logs that should be excluded from the Collector Service pipeline,
    // all other logs should be included.
    // If both Include and Exclude are specified, Include filtering occurs first.
    exclude: LogMatchProperties,

    // LogConditions is a list of OTTL conditions for an ottllog context.
    // If any condition resolves to true, the log event will be dropped.
    // Supports `and`, `or`, and `()`
    #[allow(dead_code)]
    log_record: Vec<String>,
}

/// LogMatchProperties specifies the set of properties in a log to match against and the type of string pattern matching to use.
#[derive(Debug, Clone, Deserialize)]
pub struct LogMatchProperties {
    // LogMatchType specifies the type of matching desired
    match_type: LogMatchType,

    // ResourceAttributes defines a list of possible resource attributes to match logs against.
    // A match occurs if any resource attribute matches all expressions in this given list.
    resource_attributes: Vec<KeyValue>,

    // RecordAttributes defines a list of possible record attributes to match logs against.
    // A match occurs if any record attribute matches at least one expression in this given list.
    record_attributes: Vec<KeyValue>,

    // SeverityTexts is a list of strings that the LogRecord's severity text field must match
    // against.
    severity_texts: Vec<String>,

    // SeverityNumberProperties defines how to match against a log record's SeverityNumber, if defined.
    severity_number: Option<LogSeverityNumberMatchProperties>,

    // LogBodies is a list of values that the LogRecord's body field must match
    // against.
    bodies: Vec<AnyValue>,
}

/// LogSeverityNumberMatchProperties specifies the requirements needed to match on the log severity field
#[derive(Debug, Clone, Deserialize)]
pub struct LogSeverityNumberMatchProperties {
    // Min is the minimum severity needed for the log record to match.
    // This corresponds to the short names specified here:
    // https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/logs/data-model.md#displaying-severity
    // this field is case-insensitive ("INFO" == "info")
    min: i32,
    // MatchUndefined lets logs records with "unknown" severity match.
    // If MinSeverity is not set, this field is ignored, as fields are not matched based on severity.
    match_undefined: bool,
}
```

Future PR
- add trace and metric support

Part of #1213

---------

Co-authored-by: Joshua MacDonald <[email protected]>
Co-authored-by: Utkarsh Umesan Pillai <[email protected]>

Author: 3 people

rust/otap-dataflow/configs/fake-filter-debug-noop.yaml

@@ -0,0 +1,71 @@
+settings:
+  default_pipeline_ctrl_msg_channel_size: 100
+  default_node_ctrl_msg_channel_size: 100
+  default_pdata_channel_size: 100
+
+nodes:
+  receiver:
+    kind: receiver
+    plugin_urn: "urn:otel:otap:fake_data_generator:receiver"
+    out_ports:
+      out_port:
+        destinations:
+          - filter
+        dispatch_strategy: round_robin
+    config:
+      traffic_config:
+        max_signal_count: 1000
+        max_batch_size: 100
+        signals_per_second: 100
+        log_weight: 100
+      registry_path: https://github.com/open-telemetry/semantic-conventions.git[model]
+  filter:
+    kind: processor
+    plugin_urn: "urn:otel:filter:processor"
+    out_ports:
+      out_port:
+        destinations:
+          - debug
+        dispatch_strategy: round_robin
+    config:
+      logs:
+        exclude:
+          match_type: strict
+          resource_attributes:
+            - key: fake_data_generator
+              value: v0
+          record_attributes:
+            - key: fake_data_generator
+              value: v0
+          severity_texts:
+            - WARN
+          severity_number: null
+          bodies:
+            - body_text
+        include:
+          match_type: strict
+          resource_attributes:
+            - key: fake_data_generator
+              value: v1
+          record_attributes:
+            - key: gen_ai.system
+              value: openai
+          severity_texts: []
+          severity_number: null
+          bodies: []
+        log_record: []
+  debug:
+    kind: processor
+    plugin_urn: "urn:otel:debug:processor"
+    out_ports:
+      out_port:
+        destinations:
+          - noop
+        dispatch_strategy: round_robin
+    config:
+      verbosity: detailed
+      mode: signal
+  noop:
+    kind: exporter
+    plugin_urn: "urn:otel:noop:exporter"
+    config:

rust/otap-dataflow/crates/otap/src/fake_data_generator/fake_signal.rs

@@ -11,7 +11,7 @@ use crate::fake_data_generator::fake_data::{
     current_time, delay, gen_span_id, gen_trace_id, get_scope_name, get_scope_version,
 };
 use otel_arrow_rust::proto::opentelemetry::{
-    common::v1::{AnyValue, InstrumentationScope},
+    common::v1::{AnyValue, InstrumentationScope, KeyValue},
     logs::v1::{LogRecord, LogsData, ResourceLogs, ScopeLogs, SeverityNumber},
     metrics::v1::{
         AggregationTemporality, Gauge, Histogram, HistogramDataPoint, Metric, MetricsData,
@@ -37,9 +37,12 @@ pub fn fake_otlp_traces(signal_count: usize, registry: &ResolvedRegistry) -> Tra
     ];
 
     let resources: Vec<ResourceSpans> = vec![
-        ResourceSpans::build(Resource::default())
-            .scope_spans(scopes)
-            .finish(),
+        ResourceSpans::build(Resource::build(vec![KeyValue::new(
+            "fake_data_generator",
+            AnyValue::new_string("v1"),
+        )]))
+        .scope_spans(scopes)
+        .finish(),
     ];
     TracesData::new(resources)
 }
@@ -58,9 +61,12 @@ pub fn fake_otlp_logs(signal_count: usize, registry: &ResolvedRegistry) -> LogsD
     ];
 
     let resources: Vec<ResourceLogs> = vec![
-        ResourceLogs::build(Resource::default())
-            .scope_logs(scopes)
-            .finish(),
+        ResourceLogs::build(Resource::build(vec![KeyValue::new(
+            "fake_data_generator",
+            AnyValue::new_string("v1"),
+        )]))
+        .scope_logs(scopes)
+        .finish(),
     ];
 
     LogsData::new(resources)
@@ -80,9 +86,12 @@ pub fn fake_otlp_metrics(signal_count: usize, registry: &ResolvedRegistry) -> Me
     ];
 
     let resources: Vec<ResourceMetrics> = vec![
-        ResourceMetrics::build(Resource::default())
-            .scope_metrics(scopes)
-            .finish(),
+        ResourceMetrics::build(Resource::build(vec![KeyValue::new(
+            "fake_data_generator",
+            AnyValue::new_string("v1"),
+        )]))
+        .scope_metrics(scopes)
+        .finish(),
     ];
 
     MetricsData::new(resources)