[otap-dataflow ] support additional validation methods in the validation framework (open-telemetry#2027)

# Change Summary

Updated the validation exporter to perform different validation checks
(equivalence, signal drop, attribute, batch size)

- users can configure the validation framework to perform a list of
validation checks for their testing
- Defined ValidationKind enum to describe various validations to make

1. ValidationKind::SignalDrop -> checks for any change in signal count
2. ValidationKind::Attribute -> checks for existence of key or key
values and can check for nonexistent keys
3. ValidationKind::Batch -> check for correct batch sizes
4. ValidationKind::Equivalence -> checks that message hasn't been
transformed


## What issue does this PR close?

related to open-telemetry#2008 

## How are these changes tested?

Added tests for each of the new validation check methods.
Added example validation framework tests for filter processor and
attribute processor pipelines

## Are there any user-facing changes?
No

Author: c1ly

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

@@ -64,8 +64,10 @@ pub enum AnyValue {
 /// struct that represents attributes and other key value pairs
 #[derive(Debug, Clone, PartialEq, Deserialize)]
 pub struct KeyValue {
-    key: String,
-    value: AnyValue,
+    /// Attribute key.
+    pub key: String,
+    /// Attribute value.
+    pub value: AnyValue,
 }
 
 impl KeyValue {

rust/otap-dataflow/crates/pdata/src/testing/equiv/canonical.rs

@@ -161,3 +161,38 @@ pub(crate) fn assert_equivalent<T, F, G>(
         message_name
     );
 }
+
+/// Generic equivalence assertion for any OTLP signal type.
+pub(crate) fn validate_equivalent<T, F, G>(
+    left: &[T],
+    right: &[T],
+    split_fn: F,
+    canonicalize_fn: G,
+) -> bool
+where
+    T: Message + Clone + std::fmt::Debug + PartialEq,
+    F: Fn(&T) -> Vec<T>,
+    G: Fn(&mut T),
+{
+    // Split into singletons from all messages in the slices
+    let mut left_singletons: Vec<T> = left.iter().flat_map(&split_fn).collect();
+    let mut right_singletons: Vec<T> = right.iter().flat_map(&split_fn).collect();
+
+    // Canonicalize each singleton
+    for singleton in &mut left_singletons {
+        canonicalize_fn(singleton);
+    }
+    for singleton in &mut right_singletons {
+        canonicalize_fn(singleton);
+    }
+
+    // Encode to bytes and collect into BTreeSets
+    let left_set: BTreeSet<Vec<u8>> = left_singletons.iter().map(canonicalize_message).collect();
+    let right_set: BTreeSet<Vec<u8>> = right_singletons.iter().map(canonicalize_message).collect();
+
+    // Use pretty_assertions for nice diff output
+    if left_set == right_set {
+        return true;
+    }
+    false
+}

rust/otap-dataflow/crates/pdata/src/testing/equiv/logs.rs

@@ -6,7 +6,7 @@
 use crate::proto::opentelemetry::logs::v1::{LogsData, ResourceLogs, ScopeLogs};
 use crate::testing::equiv::canonical::{
     assert_equivalent, canonicalize_any_value, canonicalize_resource, canonicalize_scope,
-    canonicalize_vec,
+    canonicalize_vec, validate_equivalent,
 };
 
 /// Split a LogsData into individual singleton LogsData messages (one per log record).
@@ -73,6 +73,16 @@ pub fn assert_logs_equivalent(left: &[LogsData], right: &[LogsData]) {
     );
 }
 
+/// Validate that two collections of `LogsData` instances are equivalent.
+pub fn validate_logs_equivalent(left: &[LogsData], right: &[LogsData]) -> bool {
+    validate_equivalent(
+        left,
+        right,
+        logs_split_into_singletons,
+        logs_canonicalize_singleton_in_place,
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -254,6 +264,10 @@ mod tests {
             }],
         };
 
+        assert!(validate_logs_equivalent(
+            std::slice::from_ref(&request1),
+            std::slice::from_ref(&request2)
+        ));
         assert_logs_equivalent(&[request1], &[request2]);
     }
 }

rust/otap-dataflow/crates/pdata/src/testing/equiv/metrics.rs

@@ -10,7 +10,7 @@ use crate::proto::opentelemetry::metrics::v1::{
 };
 use crate::testing::equiv::canonical::{
     assert_equivalent, canonicalize_any_value, canonicalize_resource, canonicalize_scope,
-    canonicalize_vec,
+    canonicalize_vec, validate_equivalent,
 };
 
 /// Split a MetricsData into individual singleton MetricsData messages (one per data point).
@@ -287,6 +287,16 @@ pub fn assert_metrics_equivalent(left: &[MetricsData], right: &[MetricsData]) {
     );
 }
 
+/// Validate that two collections of `MetricsData` instances are equivalent.
+pub fn validate_metrics_equivalent(left: &[MetricsData], right: &[MetricsData]) -> bool {
+    validate_equivalent(
+        left,
+        right,
+        metrics_split_into_singletons,
+        metrics_canonicalize_singleton,
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -377,6 +387,10 @@ mod tests {
             }],
         };
 
+        assert!(validate_metrics_equivalent(
+            std::slice::from_ref(&request1),
+            std::slice::from_ref(&request2)
+        ));
         assert_metrics_equivalent(&[request1], &[request2]);
     }
 }

rust/otap-dataflow/crates/pdata/src/testing/equiv/mod.rs

@@ -25,9 +25,9 @@ use crate::proto::opentelemetry::logs::v1::LogsData;
 use crate::proto::opentelemetry::metrics::v1::MetricsData;
 use crate::proto::opentelemetry::trace::v1::TracesData;
 
-use logs::assert_logs_equivalent;
-use metrics::assert_metrics_equivalent;
-use traces::assert_traces_equivalent;
+use logs::{assert_logs_equivalent, validate_logs_equivalent};
+use metrics::{assert_metrics_equivalent, validate_metrics_equivalent};
+use traces::{assert_traces_equivalent, validate_traces_equivalent};
 
 fn otap_to_otlp_logs(msg: &OtlpProtoMessage) -> LogsData {
     match msg {
@@ -70,3 +70,24 @@ pub fn assert_equivalent(left: &[OtlpProtoMessage], right: &[OtlpProtoMessage])
         ),
     }
 }
+
+/// Validate that two OTLP protocol message slices contain equivalent data.
+/// Requires the inputs to have a single signal type.
+pub fn validate_equivalent(left: &[OtlpProtoMessage], right: &[OtlpProtoMessage]) -> bool {
+    let signal_type = left.first().expect("at least one input").signal_type();
+
+    match signal_type {
+        SignalType::Logs => validate_logs_equivalent(
+            &left.iter().map(otap_to_otlp_logs).collect::<Vec<_>>(),
+            &right.iter().map(otap_to_otlp_logs).collect::<Vec<_>>(),
+        ),
+        SignalType::Metrics => validate_metrics_equivalent(
+            &left.iter().map(otap_to_otlp_metrics).collect::<Vec<_>>(),
+            &right.iter().map(otap_to_otlp_metrics).collect::<Vec<_>>(),
+        ),
+        SignalType::Traces => validate_traces_equivalent(
+            &left.iter().map(otap_to_otlp_traces).collect::<Vec<_>>(),
+            &right.iter().map(otap_to_otlp_traces).collect::<Vec<_>>(),
+        ),
+    }
+}

rust/otap-dataflow/crates/pdata/src/testing/equiv/traces.rs

@@ -6,7 +6,7 @@
 use crate::proto::opentelemetry::trace::v1::{ResourceSpans, ScopeSpans, Status, TracesData};
 use crate::testing::equiv::canonical::{
     assert_equivalent, canonicalize_any_value, canonicalize_idvec, canonicalize_resource,
-    canonicalize_scope, canonicalize_vec,
+    canonicalize_scope, canonicalize_vec, validate_equivalent,
 };
 
 /// Split a TracesData into individual singleton TracesData messages (one per span).
@@ -96,6 +96,16 @@ pub fn assert_traces_equivalent(left: &[TracesData], right: &[TracesData]) {
     );
 }
 
+/// Validate that two collections of `TracesData` instances are equivalent.
+pub fn validate_traces_equivalent(left: &[TracesData], right: &[TracesData]) -> bool {
+    validate_equivalent(
+        left,
+        right,
+        traces_split_into_singletons,
+        traces_canonicalize_singleton_in_place,
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -278,6 +288,10 @@ mod tests {
             }],
         };
 
+        assert!(validate_traces_equivalent(
+            std::slice::from_ref(&request1),
+            std::slice::from_ref(&request2)
+        ));
         assert_traces_equivalent(&[request1], &[request2]);
     }
 }

rust/otap-dataflow/crates/validation/Cargo.toml

@@ -21,6 +21,7 @@ serde_json = { workspace = true }
 tokio-util = { workspace = true }
 tokio = { workspace = true }
 portpicker = { workspace = true }
+humantime-serde = { workspace = true }
 
 minijinja = "1"
 reqwest = { version = "0.12", features = ["json"] }
@@ -39,6 +40,7 @@ otap-df-telemetry = { path = "../telemetry" }
 otap-df-telemetry-macros = { path = "../telemetry-macros" }
 otap-df-otap = { path = "../otap" }
 otap-df-controller = { path = "../controller" }
+prost = { workspace = true }
 
 [dev-dependencies]
 tempfile = "3"