Merge branch 'main' into cijothomas/metric-api-instrument-builder-sync

Author: cijothomas

opentelemetry-jaeger/src/exporter/config/collector/http_client.rs

@@ -129,6 +129,7 @@ impl CollectorHttpClient {
 }
 
 #[cfg(test)]
+#[allow(dead_code)]
 pub(crate) mod test_http_client {
     use async_trait::async_trait;
     use bytes::Bytes;

opentelemetry-jaeger/src/exporter/thrift/agent.rs

@@ -3,7 +3,7 @@
 
 #![allow(unused_imports)]
 #![allow(unused_extern_crates)]
-#![cfg_attr(clippy, allow(clippy::too_many_arguments, clippy::type_complexity))]
+#![allow(clippy::too_many_arguments, clippy::type_complexity)]
 #![cfg_attr(rustfmt, rustfmt_skip)]
 
 extern crate thrift;

opentelemetry-jaeger/src/exporter/thrift/jaeger.rs

@@ -3,7 +3,7 @@
 
 #![allow(unused_imports)]
 #![allow(unused_extern_crates)]
-#![cfg_attr(clippy, allow(clippy::too_many_arguments, clippy::type_complexity))]
+#![allow(clippy::too_many_arguments, clippy::type_complexity)]
 #![cfg_attr(rustfmt, rustfmt_skip)]
 
 extern crate thrift;

opentelemetry-jaeger/src/exporter/thrift/zipkincore.rs

@@ -3,7 +3,7 @@
 
 #![allow(unused_imports)]
 #![allow(unused_extern_crates)]
-#![cfg_attr(clippy, allow(clippy::too_many_arguments, clippy::type_complexity))]
+#![allow(clippy::too_many_arguments, clippy::type_complexity)]
 #![cfg_attr(rustfmt, rustfmt_skip)]
 
 extern crate thrift;

opentelemetry-otlp/examples/basic-otlp-http/src/main.rs

@@ -74,8 +74,20 @@ static COMMON_ATTRIBUTES: Lazy<[KeyValue; 4]> = Lazy::new(|| {
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
-    let _ = init_tracer()?;
-    let _ = init_metrics()?;
+    let result = init_tracer();
+    assert!(
+        result.is_ok(),
+        "Init tracer failed with error: {:?}",
+        result.err()
+    );
+
+    let result = init_metrics();
+    assert!(
+        result.is_ok(),
+        "Init metrics failed with error: {:?}",
+        result.err()
+    );
+
     // Opentelemetry will not provide a global API to manage the logger provider. Application users must manage the lifecycle of the logger provider on their own. Dropping logger providers will disable log emitting.
     let logger_provider = init_logs().unwrap();
 

opentelemetry-otlp/examples/basic-otlp/src/main.rs

@@ -89,8 +89,20 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     // By binding the result to an unused variable, the lifetime of the variable
     // matches the containing block, reporting traces and metrics during the whole
     // execution.
-    let _ = init_tracer()?;
-    let _ = init_metrics()?;
+
+    let result = init_tracer();
+    assert!(
+        result.is_ok(),
+        "Init tracer failed with error: {:?}",
+        result.err()
+    );
+
+    let result = init_metrics();
+    assert!(
+        result.is_ok(),
+        "Init metrics failed with error: {:?}",
+        result.err()
+    );
 
     // Initialize logs, which sets the global loggerprovider.
     let logger_provider = init_logs().unwrap();

opentelemetry-otlp/tests/integration_test/tests/integration_tests.rs

@@ -1,3 +1,5 @@
+#![cfg(unix)]
+
 use integration_test_runner::images::Collector;
 use std::fs::File;
 use std::os::unix::fs::PermissionsExt;

opentelemetry-otlp/tests/integration_test/tests/traces.rs

@@ -1,3 +1,5 @@
+#![cfg(unix)]
+
 use integration_test_runner::asserter::{read_spans_from_json, TraceAsserter};
 use opentelemetry::global;
 use opentelemetry::global::shutdown_tracer_provider;

opentelemetry-prometheus/tests/integration_test.rs

@@ -405,11 +405,21 @@ fn gather_and_compare(registry: prometheus::Registry, expected: String, name: &'
     let encoder = TextEncoder::new();
     let metric_families = registry.gather();
     encoder.encode(&metric_families, &mut output).unwrap();
-    let output_string = String::from_utf8(output).unwrap();
+
+    let output_string = get_platform_specific_string(String::from_utf8(output).unwrap());
 
     assert_eq!(output_string, expected, "{name}");
 }
 
+///  Returns a String which uses the platform specific new line feed character.
+fn get_platform_specific_string(input: String) -> String {
+    if cfg!(windows) {
+        input.replace('\n', "\r\n")
+    } else {
+        input
+    }
+}
+
 #[test]
 fn multiple_scopes() {
     let registry = prometheus::Registry::new();
@@ -816,7 +826,8 @@ fn gather_and_compare_multi(
     let encoder = TextEncoder::new();
     let metric_families = registry.gather();
     encoder.encode(&metric_families, &mut output).unwrap();
-    let output_string = String::from_utf8(output).unwrap();
+
+    let output_string = get_platform_specific_string(String::from_utf8(output).unwrap());
 
     assert!(
         expected.contains(&output_string),

opentelemetry-proto/tests/grpc_build.rs

@@ -18,7 +18,7 @@ const TONIC_INCLUDES: &[&str] = &["src/proto/opentelemetry-proto", "src/proto"];
 
 #[test]
 fn build_tonic() {
-    let before_build = build_content_map(TONIC_OUT_DIR);
+    let before_build = build_content_map(TONIC_OUT_DIR, false);
 
     let out_dir = TempDir::new().expect("failed to create temp dir to store the generated files");
 
@@ -95,11 +95,11 @@ fn build_tonic() {
         .compile(TONIC_PROTO_FILES, TONIC_INCLUDES)
         .expect("cannot compile protobuf using tonic");
 
-    let after_build = build_content_map(out_dir.path());
+    let after_build = build_content_map(out_dir.path(), true);
     ensure_files_are_same(before_build, after_build, TONIC_OUT_DIR);
 }
 
-fn build_content_map(path: impl AsRef<Path>) -> HashMap<String, String> {
+fn build_content_map(path: impl AsRef<Path>, normalize_line_feed: bool) -> HashMap<String, String> {
     std::fs::read_dir(path)
         .expect("cannot open dictionary of generated files")
         .flatten()
@@ -108,14 +108,28 @@ fn build_content_map(path: impl AsRef<Path>) -> HashMap<String, String> {
             let file_name = path
                 .file_name()
                 .expect("file name should always exist for generated files");
-            (
-                file_name.to_string_lossy().to_string(),
-                std::fs::read_to_string(path).expect("cannot read from existing generated file"),
-            )
+
+            let mut file_contents = std::fs::read_to_string(path.clone())
+                .expect("cannot read from existing generated file");
+
+            if normalize_line_feed {
+                file_contents = get_platform_specific_string(file_contents);
+            }
+
+            (file_name.to_string_lossy().to_string(), file_contents)
         })
         .collect()
 }
 
+///  Returns a String with the platform specific new line feed character.
+fn get_platform_specific_string(input: String) -> String {
+    if cfg!(windows) {
+        input.replace('\n', "\r\n")
+    } else {
+        input
+    }
+}
+
 fn ensure_files_are_same(
     before_build: HashMap<String, String>,
     after_build: HashMap<String, String>,

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -9,7 +9,7 @@ use futures_util::{
     future::{self, Either},
     pin_mut,
     stream::{self, FusedStream},
-    Stream, StreamExt,
+    StreamExt,
 };
 use opentelemetry::{
     global,
@@ -290,7 +290,7 @@ impl<RT: Runtime> PeriodicReaderWorker<RT> {
         true
     }
 
-    async fn run(mut self, mut messages: impl Stream<Item = Message> + Unpin + FusedStream) {
+    async fn run(mut self, mut messages: impl Unpin + FusedStream<Item = Message>) {
         while let Some(message) = messages.next().await {
             if !self.process_message(message).await {
                 break;
@@ -435,9 +435,11 @@ mod tests {
             })
             .expect("callback registration should succeed");
 
+        _ = meter_provider.force_flush();
+
         // Assert
         receiver
-            .recv_timeout(interval * 2)
+            .try_recv()
             .expect("message should be available in channel, indicating a collection occurred");
     }
 

opentelemetry-sdk/src/metrics/pipeline.rs

@@ -165,9 +165,9 @@ impl SdkProducer for Pipeline {
                             // previous aggregation was of a different type
                             prev_agg.data = data;
                         }
-                        prev_agg.name = inst.name.clone();
-                        prev_agg.description = inst.description.clone();
-                        prev_agg.unit = inst.unit.clone();
+                        prev_agg.name.clone_from(&inst.name);
+                        prev_agg.description.clone_from(&inst.description);
+                        prev_agg.unit.clone_from(&inst.unit);
                     }
                     _ => continue,
                 }

opentelemetry-sdk/src/trace/span_processor.rs

@@ -42,7 +42,7 @@ use futures_util::{
     future::{self, BoxFuture, Either},
     select,
     stream::{self, FusedStream, FuturesUnordered},
-    Stream, StreamExt as _,
+    StreamExt as _,
 };
 use opentelemetry::global;
 use opentelemetry::{
@@ -399,7 +399,7 @@ impl<R: RuntimeChannel> BatchSpanProcessorInternal<R> {
         })
     }
 
-    async fn run(mut self, mut messages: impl Stream<Item = BatchMessage> + Unpin + FusedStream) {
+    async fn run(mut self, mut messages: impl Unpin + FusedStream<Item = BatchMessage>) {
         loop {
             select! {
                 // FuturesUnordered implements Fuse intelligently such that it