Merge branch 'main' into add-dependabot

Author: lalitb

.gitignore

@@ -4,3 +4,5 @@
 **/*.rs.bk
 Cargo.lock
 /.idea/
+
+.cosine

opentelemetry-http/Cargo.toml

@@ -1,10 +1,10 @@
 [package]
 name = "opentelemetry-http"
 version = "0.10.0"
-description = "Helper implementations for exchange of traces and metrics over HTTP"
+description = "Helper implementations for sending HTTP requests. Uses include propagating and extracting context over http, exporting telemetry, requesting sampling strategies."
 homepage = "https://github.com/open-telemetry/opentelemetry-rust"
 repository = "https://github.com/open-telemetry/opentelemetry-rust"
-keywords = ["opentelemetry", "tracing", "metrics"]
+keywords = ["opentelemetry", "tracing", "context", "propagation"]
 license = "Apache-2.0"
 edition = "2021"
 rust-version = "1.65"

opentelemetry-http/src/lib.rs

@@ -7,6 +7,10 @@ pub use bytes::Bytes;
 pub use http::{Request, Response};
 use opentelemetry::propagation::{Extractor, Injector};
 
+/// Helper for injecting headers into HTTP Requests. This is used for OpenTelemetry context
+/// propagation over HTTP.
+/// See [this](https://github.com/open-telemetry/opentelemetry-rust/blob/main/examples/tracing-http-propagator/README.md)
+/// for example usage.
 pub struct HeaderInjector<'a>(pub &'a mut http::HeaderMap);
 
 impl<'a> Injector for HeaderInjector<'a> {
@@ -20,6 +24,10 @@ impl<'a> Injector for HeaderInjector<'a> {
     }
 }
 
+/// Helper for extracting headers from HTTP Requests. This is used for OpenTelemetry context
+/// propagation over HTTP.
+/// See [this](https://github.com/open-telemetry/opentelemetry-rust/blob/main/examples/tracing-http-propagator/README.md)
+/// for example usage.
 pub struct HeaderExtractor<'a>(pub &'a http::HeaderMap);
 
 impl<'a> Extractor for HeaderExtractor<'a> {
@@ -39,7 +47,9 @@ impl<'a> Extractor for HeaderExtractor<'a> {
 
 pub type HttpError = Box<dyn std::error::Error + Send + Sync + 'static>;
 
-/// A minimal interface necessary for export spans over HTTP.
+/// A minimal interface necessary for sending requests over HTTP.
+/// Used primarily for exporting telemetry over HTTP. Also used for fetching
+/// sampling strategies for JaegerRemoteSampler
 ///
 /// Users sometime choose HTTP clients that relay on a certain async runtime. This trait allows
 /// users to bring their choice of HTTP client.

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

@@ -188,6 +188,8 @@ mod collector_client_tests {
     use opentelemetry::trace::TraceError;
     use opentelemetry_sdk::runtime::Tokio;
 
+    // Ignore this test as it is flaky and the opentelemetry-jaeger is on-track for deprecation
+    #[ignore]
     #[test]
     fn test_bring_your_own_client() -> Result<(), TraceError> {
         let invalid_uri_builder = new_collector_pipeline()

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

@@ -571,6 +571,8 @@ mod tests {
         assert!(valid_uri.is_ok());
     }
 
+    // Ignore this test as it is flaky and the opentelemetry-jaeger is on-track for deprecation
+    #[ignore]
     #[test]
     fn test_collector_exporter() {
         let exporter = new_collector_pipeline()

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -409,20 +409,52 @@ impl MetricReader for PeriodicReader {
 mod tests {
     use super::PeriodicReader;
     use crate::{
-        metrics::data::ResourceMetrics, metrics::reader::MetricReader, runtime,
-        testing::metrics::InMemoryMetricsExporter, Resource,
+        metrics::data::ResourceMetrics, metrics::reader::MetricReader, metrics::SdkMeterProvider,
+        runtime, testing::metrics::InMemoryMetricsExporter, Resource,
     };
+    use opentelemetry::metrics::MeterProvider;
+    use std::sync::mpsc;
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn registration_triggers_collection() {
+        // Arrange
+        let interval = std::time::Duration::from_millis(1);
+        let exporter = InMemoryMetricsExporter::default();
+        let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio)
+            .with_interval(interval)
+            .build();
+        let (sender, receiver) = mpsc::channel();
+
+        // Act
+        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
+        let meter = meter_provider.meter("test");
+        let counter = meter.u64_observable_counter("testcounter").init();
+        meter
+            .register_callback(&[counter.as_any()], move |_| {
+                sender.send(()).expect("channel should still be open");
+            })
+            .expect("callback registration should succeed");
+
+        // Assert
+        receiver
+            .recv_timeout(interval * 2)
+            .expect("message should be available in channel, indicating a collection occurred");
+    }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn unregistered_collect() {
+        // Arrange
         let exporter = InMemoryMetricsExporter::default();
         let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
-
         let mut rm = ResourceMetrics {
             resource: Resource::empty(),
             scope_metrics: Vec::new(),
         };
+
+        // Act
         let result = reader.collect(&mut rm);
+
+        // Assert
         result.expect_err("error expected when reader is not registered");
     }
 }

opentelemetry-sdk/src/propagation/trace_context.rs

@@ -97,9 +97,12 @@ impl TraceContextPropagator {
         // supported sampling bit.
         let trace_flags = TraceFlags::new(opts) & TraceFlags::SAMPLED;
 
-        let trace_state: TraceState =
-            TraceState::from_str(extractor.get(TRACESTATE_HEADER).unwrap_or(""))
-                .unwrap_or_else(|_| TraceState::default());
+        let trace_state = match extractor.get(TRACESTATE_HEADER) {
+            Some(trace_state_str) => {
+                TraceState::from_str(trace_state_str).unwrap_or_else(|_| TraceState::default())
+            }
+            None => TraceState::default(),
+        };
 
         // create context
         let span_context = SpanContext::new(trace_id, span_id, trace_flags, true, trace_state);

stress/src/logs.rs

@@ -1,9 +1,5 @@
-use opentelemetry::KeyValue;
 use opentelemetry_appender_tracing::layer;
-use opentelemetry_sdk::{
-    logs::{Config, LogProcessor, LoggerProvider},
-    Resource,
-};
+use opentelemetry_sdk::logs::{LogProcessor, LoggerProvider};
 use tracing::error;
 use tracing_subscriber::{prelude::*, Layer};
 
@@ -65,12 +61,6 @@ impl LogProcessor for NoOpLogProcessor {
 fn main() {
     // LoggerProvider with a no-op processor.
     let provider: LoggerProvider = LoggerProvider::builder()
-        .with_config(
-            Config::default().with_resource(Resource::new(vec![KeyValue::new(
-                "service.name",
-                "log-appender-tracing-example",
-            )])),
-        )
         .with_log_processor(NoOpLogProcessor {})
         .build();
 

stress/src/traces.rs

@@ -1,19 +1,44 @@
 use lazy_static::lazy_static;
 use opentelemetry::{
-    trace::{Span, SpanBuilder, Tracer, TracerProvider as _},
-    KeyValue,
+    trace::{Span, SpanBuilder, TraceResult, Tracer, TracerProvider as _},
+    Context, KeyValue,
+};
+use opentelemetry_sdk::{
+    export::trace::SpanData,
+    trace::{self as sdktrace, SpanProcessor},
 };
-use opentelemetry_sdk::trace as sdktrace;
 
 mod throughput;
 
 lazy_static! {
     static ref PROVIDER: sdktrace::TracerProvider = sdktrace::TracerProvider::builder()
         .with_config(sdktrace::config().with_sampler(sdktrace::Sampler::AlwaysOn))
+        .with_span_processor(NoOpSpanProcessor {})
         .build();
     static ref TRACER: sdktrace::Tracer = PROVIDER.tracer("stress");
 }
 
+#[derive(Debug)]
+pub struct NoOpSpanProcessor;
+
+impl SpanProcessor for NoOpSpanProcessor {
+    fn on_start(&self, _span: &mut opentelemetry_sdk::trace::Span, _cx: &Context) {
+        // No-op
+    }
+
+    fn on_end(&self, _span: SpanData) {
+        // No-op
+    }
+
+    fn force_flush(&self) -> TraceResult<()> {
+        Ok(())
+    }
+
+    fn shutdown(&mut self) -> TraceResult<()> {
+        Ok(())
+    }
+}
+
 fn main() {
     throughput::test_throughput(test_span);
 }