Merge branch 'main' into otel-json

opentelemetry-prometheus/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## vNext
 
+### Added
+
+- Add `ResourceSelector` to allow attaching resource as attributes to metrics [#1608](https://github.com/open-telemetry/opentelemetry-rust/pull/1608)
+
 ## v0.15.0
 
 ### Changed

opentelemetry-prometheus/src/config.rs

@@ -7,7 +7,7 @@ use opentelemetry_sdk::metrics::{
 };
 use std::sync::{Arc, Mutex};
 
-use crate::{Collector, PrometheusExporter};
+use crate::{Collector, PrometheusExporter, ResourceSelector};
 
 /// [PrometheusExporter] configuration options
 #[derive(Default)]
@@ -19,6 +19,7 @@ pub struct ExporterBuilder {
     namespace: Option<String>,
     disable_scope_info: bool,
     reader: ManualReaderBuilder,
+    resource_selector: ResourceSelector,
 }
 
 impl fmt::Debug for ExporterBuilder {
@@ -114,6 +115,19 @@ impl ExporterBuilder {
         self
     }
 
+    /// Configures whether to export resource as attributes with every metric.
+    ///
+    /// Note that this is orthogonal to the `target_info` metric, which can be disabled using `without_target_info`.
+    ///
+    /// If you called `without_target_info` and `with_resource_selector` with `ResourceSelector::None`, resource will not be exported at all.
+    pub fn with_resource_selector(
+        mut self,
+        resource_selector: impl Into<ResourceSelector>,
+    ) -> Self {
+        self.resource_selector = resource_selector.into();
+        self
+    }
+
     /// Registers an external [MetricProducer] with this reader.
     ///
     /// The producer is used as a source of aggregated metric data which is
@@ -136,6 +150,8 @@ impl ExporterBuilder {
             create_target_info_once: OnceCell::new(),
             namespace: self.namespace,
             inner: Mutex::new(Default::default()),
+            resource_selector: self.resource_selector,
+            resource_labels_once: OnceCell::new(),
         };
 
         let registry = self.registry.unwrap_or_default();

opentelemetry-prometheus/src/lib.rs

@@ -132,9 +132,11 @@ const SCOPE_INFO_KEYS: [&str; 2] = ["otel_scope_name", "otel_scope_version"];
 const COUNTER_SUFFIX: &str = "_total";
 
 mod config;
+mod resource_selector;
 mod utils;
 
 pub use config::ExporterBuilder;
+pub use resource_selector::ResourceSelector;
 
 /// Creates a builder to configure a [PrometheusExporter]
 pub fn exporter() -> ExporterBuilder {
@@ -186,8 +188,10 @@ struct Collector {
     without_counter_suffixes: bool,
     disable_scope_info: bool,
     create_target_info_once: OnceCell<MetricFamily>,
+    resource_labels_once: OnceCell<Vec<LabelPair>>,
     namespace: Option<String>,
     inner: Mutex<CollectorInner>,
+    resource_selector: ResourceSelector,
 }
 
 #[derive(Default)]
@@ -299,10 +303,15 @@ impl prometheus::core::Collector for Collector {
             // Resource should be immutable, we don't need to compute again
             create_info_metric(TARGET_INFO_NAME, TARGET_INFO_DESCRIPTION, &metrics.resource)
         });
+
         if !self.disable_target_info && !metrics.resource.is_empty() {
             res.push(target_info.clone())
         }
 
+        let resource_labels = self
+            .resource_labels_once
+            .get_or_init(|| self.resource_selector.select(&metrics.resource));
+
         for scope_metrics in metrics.scope_metrics {
             let scope_labels = if !self.disable_scope_info {
                 if !scope_metrics.scope.attributes.is_empty() {
@@ -326,6 +335,9 @@ impl prometheus::core::Collector for Collector {
                     labels.push(l_version);
                 }
 
+                if !resource_labels.is_empty() {
+                    labels.extend(resource_labels.iter().cloned());
+                }
                 labels
             } else {
                 Vec::new()

opentelemetry-prometheus/src/resource_selector.rs

@@ -0,0 +1,43 @@
+use crate::get_attrs;
+use opentelemetry::Key;
+use opentelemetry_sdk::Resource;
+use prometheus::proto::LabelPair;
+use std::collections::HashSet;
+
+/// `ResourceSelector` is used to select which resource to export with every metrics.
+///
+/// By default, the exporter will only export resource as `target_info` metrics but not inline in every
+/// metrics. You can disable this behavior by calling [`without_target_info`](crate::ExporterBuilder::without_target_info)
+///
+/// You can add resource to every metrics by set `ResourceSelector` to anything other than `None`.
+///
+/// By default, ResourceSelector is `None`, meaning resource will not be attributes of every metrics.
+#[derive(Debug, Default)]
+#[non_exhaustive]
+pub enum ResourceSelector {
+    /// Export all resource attributes with every metrics.
+    All,
+    /// Do not export any resource attributes with every metrics.
+    #[default]
+    None,
+    /// Export only the resource attributes in the allow list with every metrics.
+    KeyAllowList(HashSet<Key>),
+}
+
+impl From<HashSet<Key>> for ResourceSelector {
+    fn from(keys: HashSet<Key>) -> Self {
+        ResourceSelector::KeyAllowList(keys)
+    }
+}
+
+impl ResourceSelector {
+    pub(crate) fn select(&self, resource: &Resource) -> Vec<LabelPair> {
+        match self {
+            ResourceSelector::All => get_attrs(&mut resource.iter(), &[]),
+            ResourceSelector::None => Vec::new(),
+            ResourceSelector::KeyAllowList(keys) => {
+                get_attrs(&mut resource.iter().filter(|(k, _)| keys.contains(k)), &[])
+            }
+        }
+    }
+}

opentelemetry-prometheus/tests/data/resource_in_every_metrics.txt

@@ -0,0 +1,9 @@
+# HELP bar_ratio a fun little gauge
+# TYPE bar_ratio gauge
+bar_ratio{A="B",C="D",otel_scope_name="testmeter",otel_scope_version="v0.1.0",service_name="prometheus_test",telemetry_sdk_language="rust",telemetry_sdk_name="opentelemetry",telemetry_sdk_version="latest"} 1
+# HELP otel_scope_info Instrumentation Scope metadata
+# TYPE otel_scope_info gauge
+otel_scope_info{otel_scope_name="testmeter",otel_scope_version="v0.1.0"} 1
+# HELP target_info Target metadata
+# TYPE target_info gauge
+target_info{service_name="prometheus_test",telemetry_sdk_language="rust",telemetry_sdk_name="opentelemetry",telemetry_sdk_version="latest"} 1

opentelemetry-prometheus/tests/data/select_resource_in_every_metrics.txt

@@ -0,0 +1,9 @@
+# HELP bar_ratio a fun little gauge
+# TYPE bar_ratio gauge
+bar_ratio{A="B",C="D",otel_scope_name="testmeter",otel_scope_version="v0.1.0",service_name="prometheus_test"} 1
+# HELP otel_scope_info Instrumentation Scope metadata
+# TYPE otel_scope_info gauge
+otel_scope_info{otel_scope_name="testmeter",otel_scope_version="v0.1.0"} 1
+# HELP target_info Target metadata
+# TYPE target_info gauge
+target_info{service_name="prometheus_test",telemetry_sdk_language="rust",telemetry_sdk_name="opentelemetry",telemetry_sdk_version="latest"} 1

opentelemetry-prometheus/tests/integration_test.rs

@@ -1,11 +1,12 @@
+use std::collections::HashSet;
 use std::fs;
 use std::path::Path;
 use std::time::Duration;
 
 use opentelemetry::metrics::{Meter, MeterProvider as _, Unit};
 use opentelemetry::Key;
 use opentelemetry::KeyValue;
-use opentelemetry_prometheus::ExporterBuilder;
+use opentelemetry_prometheus::{ExporterBuilder, ResourceSelector};
 use opentelemetry_sdk::metrics::{new_view, Aggregation, Instrument, SdkMeterProvider, Stream};
 use opentelemetry_sdk::resource::{
     EnvResourceDetector, SdkProvidedResourceDetector, TelemetryResourceDetector,
@@ -307,6 +308,39 @@ fn prometheus_exporter_integration() {
             }),
             ..Default::default()
         },
+        TestCase {
+            name: "with resource in every metrics",
+            builder: ExporterBuilder::default().with_resource_selector(ResourceSelector::All),
+            expected_file: "resource_in_every_metrics.txt",
+            record_metrics: Box::new(|meter| {
+                let attrs = vec![Key::new("A").string("B"), Key::new("C").string("D")];
+                let gauge = meter
+                    .i64_up_down_counter("bar")
+                    .with_description("a fun little gauge")
+                    .with_unit(Unit::new("1"))
+                    .init();
+                gauge.add(2, &attrs);
+                gauge.add(-1, &attrs);
+            }),
+            ..Default::default()
+        },
+        TestCase {
+            name: "with select resource in every metrics",
+            builder: ExporterBuilder::default()
+                .with_resource_selector(HashSet::from([Key::new("service.name")])),
+            expected_file: "select_resource_in_every_metrics.txt",
+            record_metrics: Box::new(|meter| {
+                let attrs = vec![Key::new("A").string("B"), Key::new("C").string("D")];
+                let gauge = meter
+                    .i64_up_down_counter("bar")
+                    .with_description("a fun little gauge")
+                    .with_unit(Unit::new("1"))
+                    .init();
+                gauge.add(2, &attrs);
+                gauge.add(-1, &attrs);
+            }),
+            ..Default::default()
+        },
     ];
 
     for tc in test_cases {

opentelemetry-sdk/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## vNext
 
+- Fix SimpleSpanProcessor to be consistent with log counterpart. Also removed
+  dependency on crossbeam-channel.
+  [1612](https://github.com/open-telemetry/opentelemetry-rust/pull/1612/files)
+
 ## v0.22.1
 
 ### Fixed

opentelemetry-sdk/Cargo.toml

@@ -14,7 +14,6 @@ opentelemetry = { version = "0.22", path = "../opentelemetry/" }
 opentelemetry-http = { version = "0.11", path = "../opentelemetry-http", optional = true }
 async-std = { workspace = true, features = ["unstable"], optional = true }
 async-trait = { workspace = true, optional = true }
-crossbeam-channel = { version = "0.5", optional = true }
 futures-channel = "0.3"
 futures-executor = { workspace = true }
 futures-util = { workspace = true, features = ["std", "sink", "async-await-macro"] }
@@ -45,9 +44,9 @@ pprof = { version = "0.13", features = ["flamegraph", "criterion"] }
 
 [features]
 default = ["trace"]
-trace = ["opentelemetry/trace", "crossbeam-channel", "rand", "async-trait", "percent-encoding"]
+trace = ["opentelemetry/trace", "rand", "async-trait", "percent-encoding"]
 jaeger_remote_sampler = ["trace", "opentelemetry-http", "http", "serde", "serde_json", "url"]
-logs = ["opentelemetry/logs", "crossbeam-channel", "async-trait", "serde_json"]
+logs = ["opentelemetry/logs", "async-trait", "serde_json"]
 logs_level_enabled = ["logs", "opentelemetry/logs_level_enabled"]
 metrics = ["opentelemetry/metrics", "glob", "async-trait"]
 testing = ["opentelemetry/testing", "trace", "metrics", "logs", "rt-async-std", "rt-tokio", "rt-tokio-current-thread", "tokio/macros", "tokio/rt-multi-thread"]

opentelemetry-sdk/src/logs/log_processor.rs

@@ -52,12 +52,10 @@ pub trait LogProcessor: Send + Sync + Debug {
     fn event_enabled(&self, level: Severity, target: &str, name: &str) -> bool;
 }
 
-/// A [`LogProcessor`] that exports synchronously when logs are emitted.
-///
-/// # Examples
-///
-/// Note that the simple processor exports synchronously every time a log is
-/// emitted. If you find this limiting, consider the batch processor instead.
+/// A [LogProcessor] that passes logs to the configured `LogExporter`, as soon
+/// as they are emitted, without any batching. This is typically useful for
+/// debugging and testing. For scenarios requiring higher
+/// performance/throughput, consider using [BatchLogProcessor].
 #[derive(Debug)]
 pub struct SimpleLogProcessor {
     exporter: Mutex<Box<dyn LogExporter>>,