feat(metrics): add scope_attributes to MetricSelector for view filtering

Expand MetricSelector to support scope_attributes as a new selector
dimension. When configured, the view only applies to instruments whose
scope contains all specified attribute key-value pairs.

Author: pritishnahar95

rust/otap-dataflow/crates/config/src/pipeline/telemetry/metrics/views.rs

@@ -3,6 +3,8 @@
 
 //! Metrics views level configurations.
 
+use std::collections::HashMap;
+
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 
@@ -24,6 +26,10 @@ pub struct MetricSelector {
     /// The instrumentation scope (meter) name to match.
     /// When set, the view only applies to instruments created under this scope.
     pub scope_name: Option<String>,
+    /// The instrumentation scope attributes to match.
+    /// When set, the view only applies to instruments whose scope contains all
+    /// of the specified attribute key-value pairs.
+    pub scope_attributes: Option<HashMap<String, String>>,
 }
 
 /// OpenTelemetry Metric Stream configuration.
@@ -115,4 +121,27 @@ mod tests {
             Some("exporter_sent_log_records")
         );
     }
+
+    #[test]
+    fn test_view_config_with_scope_attributes() {
+        let yaml_str = r#"
+            selector:
+              scope_name: "my.library"
+              scope_attributes:
+                env: "production"
+                region: "us-east-1"
+            stream:
+              description: "Production histograms"
+            "#;
+        let config: ViewConfig = serde_yaml::from_str(yaml_str).unwrap();
+        assert_eq!(config.selector.scope_name.as_deref(), Some("my.library"));
+        let attrs = config.selector.scope_attributes.unwrap();
+        assert_eq!(attrs.len(), 2);
+        assert_eq!(attrs.get("env").unwrap(), "production");
+        assert_eq!(attrs.get("region").unwrap(), "us-east-1");
+        assert_eq!(
+            config.stream.description.as_deref(),
+            Some("Production histograms")
+        );
+    }
 }

rust/otap-dataflow/crates/telemetry/src/otel_sdk/meter_provider/views_provider.rs

@@ -54,6 +54,18 @@ impl DeclarativeView {
                 return None;
             }
 
+            if let Some(scope_attributes) = &config.selector.scope_attributes {
+                let scope_attrs: Vec<_> = instrument.scope().attributes().collect();
+                for (key, value) in scope_attributes {
+                    let matched = scope_attrs
+                        .iter()
+                        .any(|kv| kv.key.as_str() == key && kv.value.to_string() == *value);
+                    if !matched {
+                        return None;
+                    }
+                }
+            }
+
             let mut stream_builder = Stream::builder();
             if let Some(name) = &config.stream.name {
                 stream_builder = stream_builder.with_name(name.clone());
@@ -69,6 +81,8 @@ impl DeclarativeView {
 
 #[cfg(test)]
 mod tests {
+    use std::collections::HashMap;
+
     use super::*;
     use opentelemetry_sdk::metrics::SdkMeterProvider;
     use otap_df_config::pipeline::telemetry::metrics::views::{
@@ -81,6 +95,7 @@ mod tests {
             selector: MetricSelector {
                 instrument_name: Some("requests.total".to_string()),
                 scope_name: None,
+                scope_attributes: None,
             },
             stream: MetricStream {
                 name: Some("http.requests.total".to_string()),
@@ -99,6 +114,7 @@ mod tests {
             selector: MetricSelector {
                 instrument_name: Some("requests.total".to_string()),
                 scope_name: None,
+                scope_attributes: None,
             },
             stream: MetricStream {
                 name: Some("http.requests.total".to_string()),
@@ -123,6 +139,7 @@ mod tests {
             selector: MetricSelector {
                 instrument_name: Some("requests.total".to_string()),
                 scope_name: None,
+                scope_attributes: None,
             },
             stream: MetricStream {
                 name: Some("http.requests.total".to_string()),
@@ -139,6 +156,7 @@ mod tests {
             selector: MetricSelector {
                 instrument_name: Some("requests.total".to_string()),
                 scope_name: Some("my.scope".to_string()),
+                scope_attributes: None,
             },
             stream: MetricStream {
                 name: Some("http.requests.total".to_string()),
@@ -157,6 +175,7 @@ mod tests {
             selector: MetricSelector {
                 instrument_name: None,
                 scope_name: Some("azure_monitor_exporter.metrics".to_string()),
+                scope_attributes: None,
             },
             stream: MetricStream {
                 name: None,
@@ -168,4 +187,26 @@ mod tests {
         let result = ViewsProvider::configure(sdk_meter_builder, views_config);
         assert!(result.is_ok());
     }
+
+    #[test]
+    fn test_views_provider_configure_with_scope_attributes() {
+        let view_config = ViewConfig {
+            selector: MetricSelector {
+                instrument_name: None,
+                scope_name: Some("my.library".to_string()),
+                scope_attributes: Some(HashMap::from([
+                    ("env".to_string(), "production".to_string()),
+                    ("region".to_string(), "us-east-1".to_string()),
+                ])),
+            },
+            stream: MetricStream {
+                name: None,
+                description: Some("Production histograms".to_string()),
+            },
+        };
+        let views_config = vec![view_config];
+        let sdk_meter_builder = SdkMeterProvider::builder();
+        let result = ViewsProvider::configure(sdk_meter_builder, views_config);
+        assert!(result.is_ok());
+    }
 }