initial commit

lalitb · lalitb · commit 34f5e979ff14 · 2024-05-24T10:24:49.000-07:00
diff --git a/opentelemetry-otlp/src/exporter/http/mod.rs b/opentelemetry-otlp/src/exporter/http/mod.rs
@@ -307,12 +307,16 @@ impl OtlpHttpClient {
     fn build_trace_export_body(
         &self,
         spans: Vec<SpanData>,
+        resource: &opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema,
     ) -> opentelemetry::trace::TraceResult<(Vec<u8>, &'static str)> {
         use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
 
-        let req = ExportTraceServiceRequest {
-            resource_spans: spans.into_iter().map(Into::into).collect(),
-        };
+        let resource_spans = spans
+            .into_iter()
+            .map(|log_event| (log_event, resource).into())
+            .collect::<Vec<_>>();
+
+        let req = ExportTraceServiceRequest { resource_spans };
         match self.protocol {
             #[cfg(feature = "http-json")]
             Protocol::HttpJson => match serde_json::to_string_pretty(&req) {
diff --git a/opentelemetry-otlp/src/exporter/http/trace.rs b/opentelemetry-otlp/src/exporter/http/trace.rs
@@ -21,7 +21,7 @@ impl SpanExporter for OtlpHttpClient {
             Err(err) => return Box::pin(std::future::ready(Err(err))),
         };
 
-        let (body, content_type) = match self.build_trace_export_body(batch) {
+        let (body, content_type) = match self.build_trace_export_body(batch, &self.resource) {
             Ok(body) => body,
             Err(e) => return Box::pin(std::future::ready(Err(e))),
         };
@@ -66,4 +66,8 @@ impl SpanExporter for OtlpHttpClient {
     fn shutdown(&mut self) {
         let _ = self.client.lock().map(|mut c| c.take());
     }
+
+    fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {
+        self.resource = resource.into();
+    }
 }
diff --git a/opentelemetry-otlp/src/exporter/tonic/logs.rs b/opentelemetry-otlp/src/exporter/tonic/logs.rs
@@ -12,7 +12,7 @@ use super::BoxInterceptor;
 pub(crate) struct TonicLogsClient {
     inner: Option<ClientInner>,
     #[allow(dead_code)]
-    // <allow dead> would be removed once we support set_resource for metrics and traces.
+    // <allow dead> would be removed once we support set_resource for metrics.
     resource: opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema,
 }
 
diff --git a/opentelemetry-otlp/src/exporter/tonic/trace.rs b/opentelemetry-otlp/src/exporter/tonic/trace.rs
@@ -12,6 +12,9 @@ use super::BoxInterceptor;
 
 pub(crate) struct TonicTracesClient {
     inner: Option<ClientInner>,
+    #[allow(dead_code)]
+    // <allow dead> would be removed once we support set_resource for metrics.
+    resource: opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema,
 }
 
 struct ClientInner {
@@ -43,6 +46,7 @@ impl TonicTracesClient {
                 client,
                 interceptor,
             }),
+            resource: Default::default(),
         }
     }
 }
@@ -66,14 +70,21 @@ impl SpanExporter for TonicTracesClient {
             }
         };
 
+        // TODO: Avoid cloning here.
+        let resource_spans = {
+            batch
+                .into_iter()
+                .map(|log_data| (log_data, &self.resource))
+                .map(Into::into)
+                .collect()
+        };
+
         Box::pin(async move {
             client
                 .export(Request::from_parts(
                     metadata,
                     extensions,
-                    ExportTraceServiceRequest {
-                        resource_spans: batch.into_iter().map(Into::into).collect(),
-                    },
+                    ExportTraceServiceRequest { resource_spans },
                 ))
                 .await
                 .map_err(crate::Error::from)?;
@@ -85,4 +96,8 @@ impl SpanExporter for TonicTracesClient {
     fn shutdown(&mut self) {
         let _ = self.inner.take();
     }
+
+    fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {
+        self.resource = resource.into();
+    }
 }
diff --git a/opentelemetry-otlp/src/span.rs b/opentelemetry-otlp/src/span.rs
@@ -227,4 +227,8 @@ impl opentelemetry_sdk::export::trace::SpanExporter for SpanExporter {
     fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {
         self.0.export(batch)
     }
+
+    fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {
+        self.0.set_resource(resource);
+    }
 }
diff --git a/opentelemetry-proto/src/transform/trace.rs b/opentelemetry-proto/src/transform/trace.rs
@@ -4,7 +4,7 @@ pub mod tonic {
     use crate::proto::tonic::trace::v1::{span, status, ResourceSpans, ScopeSpans, Span, Status};
     use crate::transform::common::{
         to_nanos,
-        tonic::{resource_attributes, Attributes},
+        tonic::{Attributes, ResourceAttributesWithSchema},
     };
     use opentelemetry::trace;
     use opentelemetry::trace::{Link, SpanId, SpanKind};
@@ -45,19 +45,15 @@ pub mod tonic {
         }
     }
 
-    impl From<SpanData> for ResourceSpans {
-        fn from(source_span: SpanData) -> Self {
+    impl From<(SpanData, &ResourceAttributesWithSchema)> for ResourceSpans {
+        fn from((source_span, resource): (SpanData, &ResourceAttributesWithSchema)) -> Self {
             let span_kind: span::SpanKind = source_span.span_kind.into();
             ResourceSpans {
                 resource: Some(Resource {
-                    attributes: resource_attributes(&source_span.resource).0,
+                    attributes: resource.attributes.0.clone(),
                     dropped_attributes_count: 0,
                 }),
-                schema_url: source_span
-                    .resource
-                    .schema_url()
-                    .map(|url| url.to_string())
-                    .unwrap_or_default(),
+                schema_url: resource.schema_url.clone().unwrap_or_default(),
                 scope_spans: vec![ScopeSpans {
                     schema_url: source_span
                         .instrumentation_lib
diff --git a/opentelemetry-stdout/src/trace/exporter.rs b/opentelemetry-stdout/src/trace/exporter.rs
@@ -5,13 +5,15 @@ use opentelemetry_sdk::export::{self, trace::ExportResult};
 use std::io::{stdout, Write};
 
 use crate::trace::transform::SpanData;
+use opentelemetry_sdk::resource::Resource;
 
 type Encoder = Box<dyn Fn(&mut dyn Write, SpanData) -> TraceResult<()> + Send + Sync>;
 
 /// An OpenTelemetry exporter that writes to stdout on export.
 pub struct SpanExporter {
     writer: Option<Box<dyn Write + Send + Sync>>,
     encoder: Encoder,
+    resource: Resource,
 }
 
 impl fmt::Debug for SpanExporter {
@@ -36,7 +38,11 @@ impl Default for SpanExporter {
 impl opentelemetry_sdk::export::trace::SpanExporter for SpanExporter {
     fn export(&mut self, batch: Vec<export::trace::SpanData>) -> BoxFuture<'static, ExportResult> {
         let res = if let Some(writer) = &mut self.writer {
-            (self.encoder)(writer, crate::trace::SpanData::from(batch)).and_then(|_| {
+            (self.encoder)(
+                writer,
+                crate::trace::SpanData::from((batch, &self.resource)),
+            )
+            .and_then(|_| {
                 writer
                     .write_all(b"\n")
                     .map_err(|err| TraceError::Other(Box::new(err)))
@@ -51,6 +57,10 @@ impl opentelemetry_sdk::export::trace::SpanExporter for SpanExporter {
     fn shutdown(&mut self) {
         self.writer.take();
     }
+
+    fn set_resource(&mut self, res: &opentelemetry_sdk::Resource) {
+        self.resource = res.clone();
+    }
 }
 
 /// Configuration for the stdout trace exporter
@@ -107,6 +117,7 @@ impl SpanExporterBuilder {
     pub fn build(self) -> SpanExporter {
         SpanExporter {
             writer: Some(self.writer.unwrap_or_else(|| Box::new(stdout()))),
+            resource: Resource::default(),
             encoder: self.encoder.unwrap_or_else(|| {
                 Box::new(|writer, spans| {
                     serde_json::to_writer(writer, &spans)
diff --git a/opentelemetry-stdout/src/trace/transform.rs b/opentelemetry-stdout/src/trace/transform.rs
@@ -9,17 +9,27 @@ pub struct SpanData {
     resource_spans: Vec<ResourceSpans>,
 }
 
-impl From<Vec<opentelemetry_sdk::export::trace::SpanData>> for SpanData {
-    fn from(sdk_spans: Vec<opentelemetry_sdk::export::trace::SpanData>) -> Self {
+impl
+    From<(
+        Vec<opentelemetry_sdk::export::trace::SpanData>,
+        &opentelemetry_sdk::Resource,
+    )> for SpanData
+{
+    fn from(
+        (sdk_spans, sdk_resource): (
+            Vec<opentelemetry_sdk::export::trace::SpanData>,
+            &opentelemetry_sdk::Resource,
+        ),
+    ) -> Self {
         let mut resource_spans = HashMap::<AttributeSet, ResourceSpans>::new();
         for sdk_span in sdk_spans {
-            let resource_schema_url = sdk_span.resource.schema_url().map(|s| s.to_string().into());
+            let resource_schema_url = sdk_resource.schema_url().map(|s| s.to_string().into());
             let schema_url = sdk_span.instrumentation_lib.schema_url.clone();
             let scope = sdk_span.instrumentation_lib.clone().into();
-            let resource = sdk_span.resource.as_ref().into();
+            let resource: Resource = sdk_resource.into();
 
             let rs = resource_spans
-                .entry(sdk_span.resource.as_ref().into())
+                .entry(sdk_resource.into())
                 .or_insert_with(move || ResourceSpans {
                     resource,
                     scope_spans: Vec::with_capacity(1),
diff --git a/opentelemetry-zipkin/src/exporter/mod.rs b/opentelemetry-zipkin/src/exporter/mod.rs
@@ -231,6 +231,8 @@ impl trace::SpanExporter for Exporter {
             self.local_endpoint.clone(),
         ))
     }
+
+    fn set_resource(&mut self, _resource: &Resource) {}
 }
 
 /// Wrap type for errors from opentelemetry zipkin
diff --git a/opentelemetry-zipkin/src/exporter/model/span.rs b/opentelemetry-zipkin/src/exporter/model/span.rs
@@ -60,9 +60,8 @@ mod tests {
     use crate::exporter::model::span::{Kind, Span};
     use crate::exporter::model::{into_zipkin_span, OTEL_ERROR_DESCRIPTION, OTEL_STATUS_CODE};
     use opentelemetry::trace::{SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId};
+    use opentelemetry_sdk::export::trace::SpanData;
     use opentelemetry_sdk::trace::{SpanEvents, SpanLinks};
-    use opentelemetry_sdk::{export::trace::SpanData, Resource};
-    use std::borrow::Cow;
     use std::collections::HashMap;
     use std::net::Ipv4Addr;
     use std::time::SystemTime;
@@ -166,7 +165,6 @@ mod tests {
                 events: SpanEvents::default(),
                 links: SpanLinks::default(),
                 status,
-                resource: Cow::Owned(Resource::default()),
                 instrumentation_lib: Default::default(),
             };
             let local_endpoint = Endpoint::new("test".into(), None);
diff --git a/stress/src/traces.rs b/stress/src/traces.rs
@@ -45,6 +45,8 @@ impl SpanProcessor for NoOpSpanProcessor {
     fn shutdown(&mut self) -> TraceResult<()> {
         Ok(())
     }
+
+    fn set_resource(&mut self, _resource: &opentelemetry_sdk::Resource) {}
 }
 
 fn main() {

Original file line number	Diff line number	Diff line change
`@@ -12,7 +12,7 @@ use super::BoxInterceptor;`
`12`	`12`	`pub(crate) struct TonicLogsClient {`
`13`	`13`	`inner: Option<ClientInner>,`
`14`	`14`	`#[allow(dead_code)]`
`15`		`- // <allow dead> would be removed once we support set_resource for metrics and traces.`
	`15`	`+ // <allow dead> would be removed once we support set_resource for metrics.`
`16`	`16`	`resource: opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema,`
`17`	`17`	`}`
`18`	`18`
Original file line number	Diff line number	Diff line change
`@@ -227,4 +227,8 @@ impl opentelemetry_sdk::export::trace::SpanExporter for SpanExporter {`
`227`	`227`	`fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, ExportResult> {`
`228`	`228`	`self.0.export(batch)`
`229`	`229`	`}`
	`230`	`+`
	`231`	`+ fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {`
	`232`	`+ self.0.set_resource(resource);`
	`233`	`+ }`
`230`	`234`	`}`
Original file line number	Diff line number	Diff line change
`@@ -231,6 +231,8 @@ impl trace::SpanExporter for Exporter {`
`231`	`231`	`self.local_endpoint.clone(),`
`232`	`232`	`))`
`233`	`233`	`}`
	`234`	`+`
	`235`	`+ fn set_resource(&mut self, _resource: &Resource) {}`
`234`	`236`	`}`
`235`	`237`
`236`	`238`	`/// Wrap type for errors from opentelemetry zipkin`
Original file line number	Diff line number	Diff line change
`@@ -45,6 +45,8 @@ impl SpanProcessor for NoOpSpanProcessor {`
`45`	`45`	`fn shutdown(&mut self) -> TraceResult<()> {`
`46`	`46`	`Ok(())`
`47`	`47`	`}`
	`48`	`+`
	`49`	`+ fn set_resource(&mut self, _resource: &opentelemetry_sdk::Resource) {}`
`48`	`50`	`}`
`49`	`51`
`50`	`52`	`fn main() {`