more changes

Author: lalitb

opentelemetry-otlp/src/exporter/http/logs.rs

@@ -19,7 +19,10 @@ impl LogExporter for OtlpHttpClient {
                 _ => Err(LogError::Other("exporter is already shut down".into())),
             })?;
 
-        let (body, content_type) = build_body(batch)?;
+        let (body, content_type) = {
+            let resource = self.resource.lock().unwrap();
+            build_body(batch, &*resource)?
+        };
         let mut request = http::Request::builder()
             .method(Method::POST)
             .uri(&self.collector_endpoint)
@@ -58,21 +61,29 @@ impl LogExporter for OtlpHttpClient {
 }
 
 #[cfg(feature = "http-proto")]
-fn build_body(logs: Vec<LogEvent>) -> LogResult<(Vec<u8>, &'static str)> {
+fn build_body(
+    logs: Vec<LogEvent>,
+    resource: &opentelemetry_sdk::Resource,
+) -> LogResult<(Vec<u8>, &'static str)> {
     use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
     use prost::Message;
+    let resource_logs = logs
+        .into_iter()
+        .map(|log_event| (log_event, resource).into())
+        .collect::<Vec<_>>();
 
-    let req = ExportLogsServiceRequest {
-        resource_logs: logs.into_iter().map(Into::into).collect(),
-    };
+    let req = ExportLogsServiceRequest { resource_logs };
     let mut buf = vec![];
     req.encode(&mut buf).map_err(crate::Error::from)?;
 
     Ok((buf, "application/x-protobuf"))
 }
 
 #[cfg(not(feature = "http-proto"))]
-fn build_body(logs: Vec<LogData>) -> LogResult<(Vec<u8>, &'static str)> {
+fn build_body(
+    logs: Vec<LogEvent>,
+    resource: &opentelemetry_sdk::Resource,
+) -> LogResult<(Vec<u8>, &'static str)> {
     Err(LogsError::Other(
         "No http protocol configured. Enable one via `http-proto`".into(),
     ))

opentelemetry-otlp/src/exporter/tonic/logs.rs

@@ -1,5 +1,7 @@
 use core::fmt;
 
+use std::sync::{Arc, Mutex};
+
 use async_trait::async_trait;
 use opentelemetry::logs::{LogError, LogResult};
 use opentelemetry_proto::tonic::collector::logs::v1::{
@@ -12,6 +14,7 @@ use super::BoxInterceptor;
 
 pub(crate) struct TonicLogsClient {
     inner: Option<ClientInner>,
+    resource: Arc<Mutex<opentelemetry_sdk::Resource>>,
 }
 
 struct ClientInner {
@@ -43,6 +46,7 @@ impl TonicLogsClient {
                 client,
                 interceptor,
             }),
+            resource: Arc::new(Mutex::new(opentelemetry_sdk::Resource::default())),
         }
     }
 }
@@ -62,13 +66,20 @@ impl LogExporter for TonicLogsClient {
             None => return Err(LogError::Other("exporter is already shut down".into())),
         };
 
+        let resource_logs = {
+            let resource = self.resource.lock().unwrap();
+            batch
+                .into_iter()
+                .map(|log_event| (log_event, &*resource))
+                .map(Into::into)
+                .collect()
+        };
+
         client
             .export(Request::from_parts(
                 metadata,
                 extensions,
-                ExportLogsServiceRequest {
-                    resource_logs: batch.into_iter().map(Into::into).collect(),
-                },
+                ExportLogsServiceRequest { resource_logs },
             ))
             .await
             .map_err(crate::Error::from)?;

opentelemetry-proto/src/transform/logs.rs

@@ -110,27 +110,35 @@ pub mod tonic {
         }
     }
 
-    impl From<opentelemetry_sdk::export::logs::LogData> for ResourceLogs {
-        fn from(log_data: opentelemetry_sdk::export::logs::LogData) -> Self {
+    impl
+        From<(
+            opentelemetry_sdk::export::logs::LogEvent,
+            &opentelemetry_sdk::Resource,
+        )> for ResourceLogs
+    {
+        fn from(
+            data: (
+                opentelemetry_sdk::export::logs::LogEvent,
+                &opentelemetry_sdk::Resource,
+            ),
+        ) -> Self {
+            let (log_event, resource) = data;
+
             ResourceLogs {
                 resource: Some(Resource {
-                    attributes: resource_attributes(&log_data.resource).0,
+                    attributes: resource_attributes(resource).0,
                     dropped_attributes_count: 0,
                 }),
-                schema_url: log_data
-                    .resource
-                    .schema_url()
-                    .map(Into::into)
-                    .unwrap_or_default(),
+                schema_url: resource.schema_url().map(Into::into).unwrap_or_default(),
                 scope_logs: vec![ScopeLogs {
-                    schema_url: log_data
+                    schema_url: log_event
                         .instrumentation
                         .schema_url
                         .clone()
                         .map(Into::into)
                         .unwrap_or_default(),
-                    scope: Some(log_data.instrumentation.into()),
-                    log_records: vec![log_data.record.into()],
+                    scope: Some(log_event.instrumentation.into()),
+                    log_records: vec![log_event.record.into()],
                 }],
             }
         }