feat(otap-dataflow): Add scope_name selector to Metric views (open-telemetry#2527)

# Change Summary

Add `scope_name` selector to Metric views
- Optional config alongside existing `instrument_name`
- Fixed yaml indentation of `fake-debug` configs

## What issue does this PR close?

* Closes open-telemetry#2500

## How are these changes tested?

Ran the fake debug configs and queried admin `/metrics` endpoint to see
resulting views

## Are there any user-facing changes?

No

Author: drewrelmas

rust/otap-dataflow/configs/fake-debug-noop-promethueus-telemetry.yaml

@@ -9,12 +9,19 @@ engine:
                 host: "0.0.0.0"
                 port: 9090
                 path: "/metrics"
-    views:
-      - selector:
-          instrument_name: "logs.produced"
-        stream:
-          name: "otlp.logs.produced.count"
-          description: "Count of logs produced"
+      views:
+        - selector:
+            scope_name: "fake_data_generator.receiver.metrics"
+            instrument_name: "logs.produced"
+          stream:
+            name: "otlp.logs.produced.count"
+            description: "Count of logs produced"
+        - selector:
+            scope_name: "debug.processor.pdata.metrics"
+            instrument_name: "logs.consumed"
+          stream:
+            name: "debug.logs.consumed.count"
+            description: "Count of log messages consumed by debug processor"
     resource:
       service.name: "fake-debug-noop-service"
 groups:
@@ -23,10 +30,10 @@ groups:
       main:
         policies:
           channel_capacity:
-              control:
-                node: 100
-                pipeline: 100
-              pdata: 128
+            control:
+              node: 100
+              pipeline: 100
+            pdata: 128
 
         nodes:
           receiver:

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

@@ -10,12 +10,19 @@ engine:
               otlp:
                 protocol: "grpc/protobuf"
                 endpoint: "http://127.0.0.1:43480"
-    views:
-      - selector:
-          instrument_name: "logs.produced"
-        stream:
-          name: "otlp.logs.produced.count"
-          description: "Count of logs produced"
+      views:
+        - selector:
+            scope_name: "fake_data_generator.receiver.metrics"
+            instrument_name: "logs.produced"
+          stream:
+            name: "otlp.logs.produced.count"
+            description: "Count of logs produced"
+        - selector:
+            scope_name: "debug.processor.pdata.metrics"
+            instrument_name: "logs.consumed"
+          stream:
+            name: "debug.logs.consumed.count"
+            description: "Count of log messages consumed by debug processor"
     resource:
       service.name: "fake-debug-noop-service"
       service.instance.id: "${env:STATEFULSET_ORDINAL:-unknown}"
@@ -25,10 +32,10 @@ groups:
       main:
         policies:
           channel_capacity:
-              control:
-                node: 100
-                pipeline: 100
-              pdata: 128
+            control:
+              node: 100
+              pipeline: 100
+            pdata: 128
 
         nodes:
           receiver:

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

@@ -21,7 +21,9 @@ pub struct ViewConfig {
 pub struct MetricSelector {
     /// The name of the instrument to match.
     pub instrument_name: Option<String>,
-    // TODO: Add more selector fields.
+    /// The instrumentation scope (meter) name to match.
+    /// When set, the view only applies to instruments created under this scope.
+    pub scope_name: Option<String>,
 }
 
 /// OpenTelemetry Metric Stream configuration.
@@ -51,6 +53,7 @@ mod tests {
             config.selector.instrument_name.as_deref(),
             Some("requests.total")
         );
+        assert_eq!(config.selector.scope_name.as_deref(), None);
         assert_eq!(config.stream.name.as_deref(), Some("http.requests.total"));
         assert_eq!(
             config.stream.description.as_deref(),
@@ -83,7 +86,33 @@ mod tests {
             "#;
         let config: ViewConfig = serde_yaml::from_str(yaml_str).unwrap();
         assert_eq!(config.selector.instrument_name.as_deref(), None);
+        assert_eq!(config.selector.scope_name.as_deref(), None);
         assert_eq!(config.stream.name.as_deref(), None);
         assert_eq!(config.stream.description.as_deref(), None);
     }
+
+    #[test]
+    fn test_view_config_with_scope_name() {
+        let yaml_str = r#"
+            selector:
+              instrument_name: "successful_rows"
+              scope_name: "azure_monitor_exporter.metrics"
+            stream:
+              name: "exporter_sent_log_records"
+              description: "Number of log records successfully sent by the exporter."
+            "#;
+        let config: ViewConfig = serde_yaml::from_str(yaml_str).unwrap();
+        assert_eq!(
+            config.selector.instrument_name.as_deref(),
+            Some("successful_rows")
+        );
+        assert_eq!(
+            config.selector.scope_name.as_deref(),
+            Some("azure_monitor_exporter.metrics")
+        );
+        assert_eq!(
+            config.stream.name.as_deref(),
+            Some("exporter_sent_log_records")
+        );
+    }
 }

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

@@ -43,6 +43,11 @@ impl DeclarativeView {
     ) -> impl Fn(&Instrument) -> Option<Stream> + Send + Sync + 'static {
         let config = self.config.clone();
         move |instrument: &Instrument| {
+            if let Some(scope_name) = &config.selector.scope_name
+                && instrument.scope().name() != scope_name
+            {
+                return None;
+            }
             if let Some(instrument_name) = &config.selector.instrument_name
                 && instrument.name() != instrument_name
             {
@@ -75,6 +80,7 @@ mod tests {
         let view_config = ViewConfig {
             selector: MetricSelector {
                 instrument_name: Some("requests.total".to_string()),
+                scope_name: None,
             },
             stream: MetricStream {
                 name: Some("http.requests.total".to_string()),
@@ -92,6 +98,7 @@ mod tests {
         let view_config = ViewConfig {
             selector: MetricSelector {
                 instrument_name: Some("requests.total".to_string()),
+                scope_name: None,
             },
             stream: MetricStream {
                 name: Some("http.requests.total".to_string()),
@@ -115,6 +122,7 @@ mod tests {
         let view_config = ViewConfig {
             selector: MetricSelector {
                 instrument_name: Some("requests.total".to_string()),
+                scope_name: None,
             },
             stream: MetricStream {
                 name: Some("http.requests.total".to_string()),
@@ -124,4 +132,40 @@ mod tests {
         let declarative_view = DeclarativeView::new(view_config);
         let _view_function = declarative_view.to_view_funtion();
     }
+
+    #[test]
+    fn test_views_provider_configure_with_scope_name() {
+        let view_config = ViewConfig {
+            selector: MetricSelector {
+                instrument_name: Some("requests.total".to_string()),
+                scope_name: Some("my.scope".to_string()),
+            },
+            stream: MetricStream {
+                name: Some("http.requests.total".to_string()),
+                description: Some("Total number of HTTP requests".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());
+    }
+
+    #[test]
+    fn test_views_provider_configure_with_scope_name_only() {
+        let view_config = ViewConfig {
+            selector: MetricSelector {
+                instrument_name: None,
+                scope_name: Some("azure_monitor_exporter.metrics".to_string()),
+            },
+            stream: MetricStream {
+                name: None,
+                description: Some("Scoped metrics".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());
+    }
 }