initial commit

opentelemetry-sdk/benches/batch_span_processor.rs

@@ -34,7 +34,6 @@ fn get_span_data() -> Vec<SpanData> {
             events: SpanEvents::default(),
             links: SpanLinks::default(),
             status: Status::Unset,
-            resource: Cow::Owned(Resource::empty()),
             instrumentation_lib: Default::default(),
         })
         .collect::<Vec<SpanData>>()

opentelemetry-sdk/benches/context.rs

@@ -140,6 +140,8 @@ impl SpanExporter for NoopExporter {
     fn export(&mut self, _spans: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
         Box::pin(futures_util::future::ready(Ok(())))
     }
+
+    fn set_resource(&mut self, _resource: &opentelemetry_sdk::Resource) {}
 }
 
 #[cfg(not(target_os = "windows"))]

opentelemetry-sdk/benches/span_builder.rs

@@ -68,6 +68,8 @@ impl SpanExporter for NoopExporter {
     fn export(&mut self, _spans: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
         Box::pin(futures_util::future::ready(Ok(())))
     }
+
+    fn set_resource(&mut self, _resource: &opentelemetry_sdk::Resource) {}
 }
 
 const MAP_KEYS: [&str; 64] = [

opentelemetry-sdk/benches/trace.rs

@@ -63,6 +63,8 @@ impl SpanExporter for VoidExporter {
     fn export(&mut self, _spans: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
         Box::pin(futures_util::future::ready(Ok(())))
     }
+
+    fn set_resource(&mut self, _resource: &opentelemetry_sdk::Resource) {}
 }
 
 fn trace_benchmark_group<F: Fn(&sdktrace::Tracer)>(c: &mut Criterion, name: &str, f: F) {

opentelemetry-sdk/src/export/trace.rs

@@ -63,6 +63,9 @@ pub trait SpanExporter: Send + Sync + Debug {
     fn force_flush(&mut self) -> BoxFuture<'static, ExportResult> {
         Box::pin(async { Ok(()) })
     }
+
+    /// Set the resource for the exporter.
+    fn set_resource(&mut self, _resource: &Resource);
 }
 
 /// `SpanData` contains all the information collected by a `Span` and can be used
@@ -92,8 +95,6 @@ pub struct SpanData {
     pub links: crate::trace::SpanLinks,
     /// Span status
     pub status: Status,
-    /// Resource contains attributes representing an entity that produced this span.
-    pub resource: Cow<'static, Resource>,
     /// Instrumentation library that produced this span
     pub instrumentation_lib: crate::InstrumentationLibrary,
 }

opentelemetry-sdk/src/testing/trace/in_memory_exporter.rs

@@ -1,4 +1,5 @@
 use crate::export::trace::{ExportResult, SpanData, SpanExporter};
+use crate::resource::Resource;
 use futures_util::future::BoxFuture;
 use opentelemetry::trace::{TraceError, TraceResult};
 use std::sync::{Arc, Mutex};
@@ -51,6 +52,7 @@ use std::sync::{Arc, Mutex};
 #[derive(Clone, Debug)]
 pub struct InMemorySpanExporter {
     spans: Arc<Mutex<Vec<SpanData>>>,
+    resource: Arc<Mutex<Resource>>,
 }
 
 impl Default for InMemorySpanExporter {
@@ -85,6 +87,7 @@ impl InMemorySpanExporterBuilder {
     pub fn build(&self) -> InMemorySpanExporter {
         InMemorySpanExporter {
             spans: Arc::new(Mutex::new(Vec::new())),
+            resource: Arc::new(Mutex::new(Resource::default())),
         }
     }
 }
@@ -142,4 +145,9 @@ impl SpanExporter for InMemorySpanExporter {
     fn shutdown(&mut self) {
         self.reset()
     }
+
+    fn set_resource(&mut self, resource: &Resource) {
+        let mut res_guard = self.resource.lock().expect("Resource lock poisoned");
+        *res_guard = resource.clone();
+    }
 }

opentelemetry-sdk/src/testing/trace/span_exporters.rs

@@ -3,7 +3,8 @@ use crate::{
         trace::{ExportResult, SpanData, SpanExporter},
         ExportError,
     },
-    trace::{Config, SpanEvents, SpanLinks},
+    resource::Resource,
+    trace::{SpanEvents, SpanLinks},
     InstrumentationLibrary,
 };
 use futures_util::future::BoxFuture;
@@ -14,7 +15,6 @@ use opentelemetry::trace::{
 use std::fmt::{Display, Formatter};
 
 pub fn new_test_export_span_data() -> SpanData {
-    let config = Config::default();
     SpanData {
         span_context: SpanContext::new(
             TraceId::from_u128(1),
@@ -33,7 +33,6 @@ pub fn new_test_export_span_data() -> SpanData {
         events: SpanEvents::default(),
         links: SpanLinks::default(),
         status: Status::Unset,
-        resource: config.resource,
         instrumentation_lib: InstrumentationLibrary::default(),
     }
 }
@@ -61,6 +60,8 @@ impl SpanExporter for TokioSpanExporter {
     fn shutdown(&mut self) {
         self.tx_shutdown.send(()).unwrap();
     }
+
+    fn set_resource(&mut self, _resource: &crate::Resource) {}
 }
 
 pub fn new_tokio_test_exporter() -> (
@@ -121,4 +122,7 @@ impl SpanExporter for NoopSpanExporter {
     fn export(&mut self, _: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
         Box::pin(std::future::ready(Ok(())))
     }
+
+    /// Set the resource for the exporter.
+    fn set_resource(&mut self, _resource: &Resource) {}
 }

opentelemetry-sdk/src/trace/provider.rs

@@ -218,11 +218,16 @@ impl Builder {
             }
         }
 
+        // Create a new vector to hold the modified processors
+        let mut processors = self.processors;
+
+        // Set the resource for each processor
+        for p in &mut processors {
+            p.set_resource(config.resource.as_ref());
+        }
+
         TracerProvider {
-            inner: Arc::new(TracerProviderInner {
-                processors: self.processors,
-                config,
-            }),
+            inner: Arc::new(TracerProviderInner { processors, config }),
         }
     }
 }
@@ -267,6 +272,10 @@ mod tests {
         fn shutdown(&mut self) -> TraceResult<()> {
             self.force_flush()
         }
+
+        fn set_resource(&mut self, _: &Resource) {
+            unimplemented!()
+        }
     }
 
     #[test]

opentelemetry-sdk/src/trace/runtime_tests.rs

@@ -2,6 +2,7 @@
 // need to run those tests one by one as the GlobalTracerProvider is a shared object between
 // threads Use cargo test -- --ignored --test-threads=1 to run those tests.
 use crate::export::trace::{ExportResult, SpanExporter};
+use crate::resource::Resource;
 #[cfg(any(feature = "rt-tokio", feature = "rt-tokio-current-thread"))]
 use crate::runtime;
 #[cfg(any(feature = "rt-tokio", feature = "rt-tokio-current-thread"))]
@@ -28,6 +29,8 @@ impl SpanExporter for SpanCountExporter {
         self.span_count.fetch_add(batch.len(), Ordering::SeqCst);
         Box::pin(async { Ok(()) })
     }
+
+    fn set_resource(&mut self, _: &Resource) {}
 }
 
 #[cfg(any(feature = "rt-tokio", feature = "rt-tokio-current-thread"))]

opentelemetry-sdk/src/trace/span.rs

@@ -9,7 +9,6 @@
 //! is possible to change its name, set its `Attributes`, and add `Links` and `Events`.
 //! These cannot be changed after the `Span`'s end time has been set.
 use crate::trace::SpanLimits;
-use crate::Resource;
 use opentelemetry::trace::{Event, Link, SpanContext, SpanId, SpanKind, Status};
 use opentelemetry::KeyValue;
 use std::borrow::Cow;
@@ -77,11 +76,11 @@ impl Span {
     /// overhead.
     pub fn exported_data(&self) -> Option<crate::export::trace::SpanData> {
         let (span_context, tracer) = (self.span_context.clone(), &self.tracer);
-        let resource = self.tracer.provider()?.config().resource.clone();
+        //let resource = self.tracer.provider()?.config().resource.clone();
 
         self.data
             .as_ref()
-            .map(|data| build_export_data(data.clone(), span_context, resource, tracer))
+            .map(|data| build_export_data(data.clone(), span_context, tracer))
     }
 }
 
@@ -225,17 +224,14 @@ impl Span {
                 processor.on_end(build_export_data(
                     data,
                     self.span_context.clone(),
-                    provider.config().resource.clone(),
                     &self.tracer,
                 ));
             }
             processors => {
-                let config = provider.config();
                 for processor in processors {
                     processor.on_end(build_export_data(
                         data.clone(),
                         self.span_context.clone(),
-                        config.resource.clone(),
                         &self.tracer,
                     ));
                 }
@@ -254,7 +250,6 @@ impl Drop for Span {
 fn build_export_data(
     data: SpanData,
     span_context: SpanContext,
-    resource: Cow<'static, Resource>,
     tracer: &crate::trace::Tracer,
 ) -> crate::export::trace::SpanData {
     crate::export::trace::SpanData {
@@ -269,7 +264,6 @@ fn build_export_data(
         events: data.events,
         links: data.links,
         status: data.status,
-        resource,
         instrumentation_lib: tracer.instrumentation_library().clone(),
     }
 }

opentelemetry-sdk/src/trace/span_processor.rs

@@ -35,6 +35,7 @@
 //! [`TracerProvider`]: opentelemetry::trace::TracerProvider
 
 use crate::export::trace::{ExportResult, SpanData, SpanExporter};
+use crate::resource::Resource;
 use crate::runtime::{RuntimeChannel, TrySend};
 use crate::trace::Span;
 use futures_channel::oneshot;
@@ -50,7 +51,7 @@ use opentelemetry::{
     Context,
 };
 use std::cmp::min;
-use std::sync::Mutex;
+use std::sync::{Arc, Mutex};
 use std::{env, fmt, str::FromStr, time::Duration};
 
 /// Delay interval between two consecutive exports.
@@ -92,6 +93,8 @@ pub trait SpanProcessor: Send + Sync + std::fmt::Debug {
     /// Shuts down the processor. Called when SDK is shut down. This is an
     /// opportunity for processors to do any cleanup required.
     fn shutdown(&mut self) -> TraceResult<()>;
+    /// Set the resource for the log processor.
+    fn set_resource(&mut self, _resource: &Resource);
 }
 
 /// A [SpanProcessor] that passes finished spans to the configured
@@ -147,6 +150,12 @@ impl SpanProcessor for SimpleSpanProcessor {
             ))
         }
     }
+
+    fn set_resource(&mut self, resource: &Resource) {
+        if let Ok(mut exporter) = self.exporter.lock() {
+            exporter.set_resource(resource);
+        }
+    }
 }
 
 /// A [`SpanProcessor`] that asynchronously buffers finished spans and reports
@@ -259,6 +268,13 @@ impl<R: RuntimeChannel> SpanProcessor for BatchSpanProcessor<R> {
             .map_err(|err| TraceError::Other(err.into()))
             .and_then(|identity| identity)
     }
+
+    fn set_resource(&mut self, resource: &Resource) {
+        let resource = Arc::new(resource.clone());
+        let _ = self
+            .message_sender
+            .try_send(BatchMessage::SetResource(resource));
+    }
 }
 
 /// Messages sent between application thread and batch span processor's work thread.
@@ -275,6 +291,8 @@ enum BatchMessage {
     Flush(Option<oneshot::Sender<ExportResult>>),
     /// Shut down the worker thread, push all spans in buffer to the backend.
     Shutdown(oneshot::Sender<ExportResult>),
+    /// Set the resource for the exporter.
+    SetResource(Arc<Resource>),
 }
 
 struct BatchSpanProcessorInternal<R> {
@@ -375,8 +393,11 @@ impl<R: RuntimeChannel> BatchSpanProcessorInternal<R> {
                 self.exporter.shutdown();
                 return false;
             }
+            // propagate the resource
+            BatchMessage::SetResource(resource) => {
+                self.exporter.set_resource(&resource);
+            }
         }
-
         true
     }
 
@@ -669,6 +690,7 @@ mod tests {
         OTEL_BSP_SCHEDULE_DELAY, OTEL_BSP_SCHEDULE_DELAY_DEFAULT,
     };
     use crate::export::trace::{ExportResult, SpanData, SpanExporter};
+    use crate::resource::Resource;
     use crate::runtime;
     use crate::testing::trace::{
         new_test_export_span_data, new_tokio_test_exporter, InMemorySpanExporterBuilder,
@@ -710,7 +732,6 @@ mod tests {
             events: SpanEvents::default(),
             links: SpanLinks::default(),
             status: Status::Unset,
-            resource: Default::default(),
             instrumentation_lib: Default::default(),
         };
         processor.on_end(unsampled);
@@ -928,6 +949,8 @@ mod tests {
             use futures_util::FutureExt;
             Box::pin((self.delay_fn)(self.delay_for).map(|_| Ok(())))
         }
+
+        fn set_resource(&mut self, _resource: &Resource) {}
     }
 
     #[test]