Merge branch 'main' into self-diagnostics

Author: lalitb

.github/workflows/semver.yml

@@ -19,4 +19,4 @@ jobs:
         with:
           components: rustfmt
       - name: cargo-semver-checks
-        uses: obi1kenobi/cargo-semver-checks-action@v2.3
+        uses: obi1kenobi/cargo-semver-checks-action@v2.4

opentelemetry-sdk/CHANGELOG.md

@@ -41,7 +41,9 @@
     - `SpanData` doesn't have the resource attributes. The `SpanExporter::export()` method needs to merge it 
       with the earlier preserved resource before export.
 
-- **Breaking** [1836](https://github.com/open-telemetry/opentelemetry-rust/pull/1836) `SpanProcessor::shutdown` now takes an immutable reference to self. Any reference can call shutdown on the processor. After the first call to `shutdown` the processor will not process any new spans. 
+- **Breaking** [1836](https://github.com/open-telemetry/opentelemetry-rust/pull/1836) `SpanProcessor::shutdown` now takes an immutable reference to self. Any reference can call shutdown on the processor. After the first call to `shutdown` the processor will not process any new spans.
+
+- **Breaking** [1850] (https://github.com/open-telemetry/opentelemetry-rust/pull/1850) `LoggerProvider::log_processors()` and `LoggerProvider::resource()` are not public methods anymore. They are only used within the `opentelemetry-sdk` crate.
 
 ## v0.23.0
 

opentelemetry-sdk/benches/attribute_set.rs

@@ -3,7 +3,7 @@ use opentelemetry::KeyValue;
 use opentelemetry_sdk::metrics::AttributeSet;
 
 // Run this benchmark with:
-// cargo bench --bench attribute_set --features=metrics
+// cargo bench --bench attribute_set
 
 fn criterion_benchmark(c: &mut Criterion) {
     attribute_set(c);

opentelemetry-sdk/benches/log.rs

@@ -1,6 +1,6 @@
-//! run with `$ cargo bench --bench log --features=logs -- --exact <test_name>` to run specific test for logs
-//! So to run test named "full-log-with-attributes/with-context" you would run `$ cargo bench --bench log --features=logs -- --exact full-log-with-attributes/with-context`
-//! To run all tests for logs you would run `$ cargo bench --bench log --features=logs`
+//! run with `$ cargo bench --bench log -- --exact <test_name>` to run specific test for logs
+//! So to run test named "full-log-with-attributes/with-context" you would run `$ cargo bench --bench log -- --exact full-log-with-attributes/with-context`
+//! To run all tests for logs you would run `$ cargo bench --bench log`
 //!
 
 use std::collections::HashMap;

opentelemetry-sdk/benches/metric_counter.rs

@@ -30,7 +30,7 @@ thread_local! {
 }
 
 // Run this benchmark with:
-// cargo bench --bench metric_counter --features=metrics
+// cargo bench --bench metric_counter
 fn create_counter() -> Counter<u64> {
     let meter_provider: SdkMeterProvider = SdkMeterProvider::builder()
         .with_reader(ManualReader::builder().build())

opentelemetry-sdk/src/logs/log_emitter.rs

@@ -90,14 +90,12 @@ impl LoggerProvider {
         Builder::default()
     }
 
-    /// Resource associated with this provider.
-    pub fn resource(&self) -> &Resource {
-        &self.inner.resource
+    pub(crate) fn log_processors(&self) -> &[Box<dyn LogProcessor>] {
+        &self.inner.processors
     }
 
-    /// Log processors associated with this provider.
-    pub fn log_processors(&self) -> &[Box<dyn LogProcessor>] {
-        &self.inner.processors
+    pub(crate) fn resource(&self) -> &Resource {
+        &self.inner.resource
     }
 
     /// Force flush all remaining logs in log processors and return results.
@@ -196,17 +194,20 @@ impl Builder {
     /// Create a new provider from this configuration.
     pub fn build(self) -> LoggerProvider {
         let resource = self.resource.unwrap_or_default();
-        // invoke set_resource on all the processors
-        for processor in &self.processors {
-            processor.set_resource(&resource);
-        }
-        LoggerProvider {
+
+        let logger_provider = LoggerProvider {
             inner: Arc::new(LoggerProviderInner {
                 processors: self.processors,
                 resource,
             }),
             is_shutdown: Arc::new(AtomicBool::new(false)),
+        };
+
+        // invoke set_resource on all the processors
+        for processor in logger_provider.log_processors() {
+            processor.set_resource(logger_provider.resource());
         }
+        logger_provider
     }
 }
 
@@ -511,7 +512,7 @@ mod tests {
 
     #[test]
     fn global_shutdown_test() {
-        // cargo test shutdown_test --features=logs
+        // cargo test global_shutdown_test --features=testing
 
         // Arrange
         let shutdown_called = Arc::new(Mutex::new(false));

opentelemetry-sdk/src/logs/log_processor.rs

@@ -536,7 +536,7 @@ mod tests {
 
     #[derive(Debug, Clone)]
     struct MockLogExporter {
-        resource: Arc<Option<Resource>>,
+        resource: Arc<Mutex<Option<Resource>>>,
     }
 
     #[async_trait]
@@ -548,15 +548,19 @@ mod tests {
         fn shutdown(&mut self) {}
 
         fn set_resource(&mut self, resource: &Resource) {
-            let res = Arc::make_mut(&mut self.resource);
-            *res = Some(resource.clone());
+            self.resource
+                .lock()
+                .map(|mut res_opt| {
+                    res_opt.replace(resource.clone());
+                })
+                .expect("mock log exporter shouldn't error when setting resource");
         }
     }
 
     // Implementation specific to the MockLogExporter, not part of the LogExporter trait
     impl MockLogExporter {
         fn get_resource(&self) -> Option<Resource> {
-            (*self.resource).clone()
+            (*self.resource).lock().unwrap().clone()
         }
     }
 
@@ -713,7 +717,7 @@ mod tests {
     #[test]
     fn test_set_resource_simple_processor() {
         let exporter = MockLogExporter {
-            resource: Arc::new(Some(Resource::default())),
+            resource: Arc::new(Mutex::new(None)),
         };
         let processor = SimpleLogProcessor::new(Box::new(exporter.clone()));
         let _ = LoggerProvider::builder()
@@ -723,15 +727,16 @@ mod tests {
                 KeyValue::new("k2", "v3"),
                 KeyValue::new("k3", "v3"),
                 KeyValue::new("k4", "v4"),
+                KeyValue::new("k5", "v5"),
             ]))
             .build();
-        assert_eq!(exporter.get_resource().unwrap().into_iter().count(), 4);
+        assert_eq!(exporter.get_resource().unwrap().into_iter().count(), 5);
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn test_set_resource_batch_processor() {
         let exporter = MockLogExporter {
-            resource: Arc::new(Some(Resource::default())),
+            resource: Arc::new(Mutex::new(None)),
         };
         let processor = BatchLogProcessor::new(
             Box::new(exporter.clone()),
@@ -745,9 +750,11 @@ mod tests {
                 KeyValue::new("k2", "v3"),
                 KeyValue::new("k3", "v3"),
                 KeyValue::new("k4", "v4"),
+                KeyValue::new("k5", "v5"),
             ]))
             .build();
-        assert_eq!(exporter.get_resource().unwrap().into_iter().count(), 4);
+        tokio::time::sleep(Duration::from_secs(2)).await; // set resource in batch span processor is not blocking. Should we make it blocking?
+        assert_eq!(exporter.get_resource().unwrap().into_iter().count(), 5);
         let _ = provider.shutdown();
     }
 

opentelemetry-sdk/src/metrics/mod.rs

@@ -153,7 +153,7 @@ mod tests {
     use std::thread;
 
     // Run all tests in this mod
-    // cargo test metrics::tests --features=metrics,testing
+    // cargo test metrics::tests --features=testing
     // Note for all tests from this point onwards in this mod:
     // "multi_thread" tokio flavor must be used else flush won't
     // be able to make progress!
@@ -190,49 +190,49 @@ mod tests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn counter_aggregation_cumulative() {
         // Run this test with stdout enabled to see output.
-        // cargo test counter_aggregation_cumulative --features=metrics,testing -- --nocapture
+        // cargo test counter_aggregation_cumulative --features=testing -- --nocapture
         counter_aggregation_helper(Temporality::Cumulative);
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn counter_aggregation_delta() {
         // Run this test with stdout enabled to see output.
-        // cargo test counter_aggregation_delta --features=metrics,testing -- --nocapture
+        // cargo test counter_aggregation_delta --features=testing -- --nocapture
         counter_aggregation_helper(Temporality::Delta);
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn histogram_aggregation_cumulative() {
         // Run this test with stdout enabled to see output.
-        // cargo test histogram_aggregation_cumulative --features=metrics,testing -- --nocapture
+        // cargo test histogram_aggregation_cumulative --features=testing -- --nocapture
         histogram_aggregation_helper(Temporality::Cumulative);
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn histogram_aggregation_delta() {
         // Run this test with stdout enabled to see output.
-        // cargo test histogram_aggregation_delta --features=metrics,testing -- --nocapture
+        // cargo test histogram_aggregation_delta --features=testing -- --nocapture
         histogram_aggregation_helper(Temporality::Delta);
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn updown_counter_aggregation_cumulative() {
         // Run this test with stdout enabled to see output.
-        // cargo test updown_counter_aggregation_cumulative --features=metrics,testing -- --nocapture
+        // cargo test updown_counter_aggregation_cumulative --features=testing -- --nocapture
         updown_counter_aggregation_helper(Temporality::Cumulative);
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn updown_counter_aggregation_delta() {
         // Run this test with stdout enabled to see output.
-        // cargo test updown_counter_aggregation_delta --features=metrics,testing -- --nocapture
+        // cargo test updown_counter_aggregation_delta --features=testing -- --nocapture
         updown_counter_aggregation_helper(Temporality::Delta);
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn gauge_aggregation() {
         // Run this test with stdout enabled to see output.
-        // cargo test gauge_aggregation --features=metrics,testing -- --nocapture
+        // cargo test gauge_aggregation --features=testing -- --nocapture
 
         // Gauge should use last value aggregation regardless of the aggregation temporality used.
         gauge_aggregation_helper(Temporality::Delta);
@@ -553,7 +553,7 @@ mod tests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn histogram_aggregation_with_invalid_aggregation_should_proceed_as_if_view_not_exist() {
         // Run this test with stdout enabled to see output.
-        // cargo test histogram_aggregation_with_invalid_aggregation_should_proceed_as_if_view_not_exist --features=metrics,testing -- --nocapture
+        // cargo test histogram_aggregation_with_invalid_aggregation_should_proceed_as_if_view_not_exist --features=testing -- --nocapture
 
         // Arrange
         let exporter = InMemoryMetricsExporter::default();
@@ -602,7 +602,7 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn spatial_aggregation_when_view_drops_attributes_observable_counter() {
-        // cargo test spatial_aggregation_when_view_drops_attributes_observable_counter --features=metrics,testing
+        // cargo test spatial_aggregation_when_view_drops_attributes_observable_counter --features=testing
 
         // Arrange
         let exporter = InMemoryMetricsExporter::default();
@@ -677,7 +677,7 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn spatial_aggregation_when_view_drops_attributes_counter() {
-        // cargo test spatial_aggregation_when_view_drops_attributes_counter --features=metrics,testing
+        // cargo test spatial_aggregation_when_view_drops_attributes_counter --features=testing
 
         // Arrange
         let exporter = InMemoryMetricsExporter::default();
@@ -754,7 +754,7 @@ mod tests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn counter_aggregation_attribute_order() {
         // Run this test with stdout enabled to see output.
-        // cargo test counter_aggregation_attribute_order --features=metrics,testing -- --nocapture
+        // cargo test counter_aggregation_attribute_order --features=testing -- --nocapture
 
         // Arrange
         let mut test_context = TestContext::new(Temporality::Delta);
@@ -987,7 +987,7 @@ mod tests {
     #[ignore = "Known bug: https://github.com/open-telemetry/opentelemetry-rust/issues/1598"]
     async fn delta_memory_efficiency_test() {
         // Run this test with stdout enabled to see output.
-        // cargo test delta_memory_efficiency_test --features=metrics,testing -- --nocapture
+        // cargo test delta_memory_efficiency_test --features=testing -- --nocapture
 
         // Arrange
         let mut test_context = TestContext::new(Temporality::Delta);
@@ -1036,7 +1036,7 @@ mod tests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn counter_multithreaded() {
         // Run this test with stdout enabled to see output.
-        // cargo test counter_multithreaded --features=metrics,testing -- --nocapture
+        // cargo test counter_multithreaded --features=testing -- --nocapture
 
         counter_multithreaded_aggregation_helper(Temporality::Delta);
         counter_multithreaded_aggregation_helper(Temporality::Cumulative);

opentelemetry-sdk/src/trace/span_processor.rs

@@ -683,7 +683,7 @@ where
 
 #[cfg(all(test, feature = "testing", feature = "trace"))]
 mod tests {
-    // cargo test trace::span_processor::tests:: --features=trace,testing
+    // cargo test trace::span_processor::tests:: --features=testing
     use super::{
         BatchSpanProcessor, SimpleSpanProcessor, SpanProcessor, OTEL_BSP_EXPORT_TIMEOUT,
         OTEL_BSP_MAX_EXPORT_BATCH_SIZE, OTEL_BSP_MAX_QUEUE_SIZE, OTEL_BSP_MAX_QUEUE_SIZE_DEFAULT,

opentelemetry-semantic-conventions/scripts/generate-consts-from-spec.sh

@@ -5,7 +5,7 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 CRATE_DIR="${SCRIPT_DIR}/../"
 
 # freeze the spec version and generator version to make generation reproducible
-SPEC_VERSION=1.25.0
+SPEC_VERSION=1.26.0
 SEMCOVGEN_VERSION=0.24.0
 
 cd "$CRATE_DIR"
@@ -53,4 +53,14 @@ fi
 # handle doc generation failures
 "${SED[@]}" 's/\[2\]\.$//' src/resource.rs src/trace.rs # remove trailing [2] from few of the doc comments
 
+# Remove the messaging.client_id definition along with its comments from the generated files
+#   - semconv "messaging.client_id" is deprecated
+#   - semconv "messaging.client.id" is to be used instead
+#   - Now because we use:
+#      	pub const {{attribute.fqn | to_const_name}}: &str = "{{attribute.fqn}}";
+#     to generate the consts, where to_const_name replaces '.' with '_', we need to remove the old definition
+#	  to avoid conflicts with the new one. Refer - https://github.com/open-telemetry/semantic-conventions/issues/1031
+"${SED[@]}" '/\/\/\/ Deprecated, use `messaging.client.id` instead\./{N;N;N;N;d;}' src/trace.rs src/resource.rs
+"${SED[@]}" '/pub const MESSAGING_CLIENT_ID: &str = "messaging.client_id";/{N;d;}' src/trace.rs src/resource.rs
+
 cargo fmt

opentelemetry-semantic-conventions/src/lib.rs

@@ -20,4 +20,4 @@ pub mod trace;
 
 /// The schema URL that matches the version of the semantic conventions that
 /// this crate defines.
-pub const SCHEMA_URL: &str = "https://opentelemetry.io/schemas/1.25.0";
+pub const SCHEMA_URL: &str = "https://opentelemetry.io/schemas/1.26.0";