[Logs SDK] Send resource once to processor and exporter, and not for every event.

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

@@ -19,7 +19,7 @@
                 _ => Err(LogError::Other("exporter is already shut down".into())),
             })?;
 
-        let (body, content_type) = build_body(batch)?;
+        let (body, content_type) = { build_body(batch, &self.resource)? };
         let mut request = http::Request::builder()
             .method(Method::POST)
             .uri(&self.collector_endpoint)
@@ -50,24 +50,36 @@
     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();
+    }
 }
 
 #[cfg(feature = "http-proto")]
-fn build_body(logs: Vec<LogData>) -> LogResult<(Vec<u8>, &'static str)> {
+fn build_body(
+    logs: Vec<LogData>,
+    resource: &opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema,
+) -> 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<LogData>,
+    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/http/mod.rs

@@ -4,6 +4,7 @@
 };
 use http::{HeaderName, HeaderValue, Uri};
 use opentelemetry_http::HttpClient;
+use opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema;
 use std::collections::HashMap;
 use std::env;
 use std::str::FromStr;
@@ -255,6 +256,9 @@
     collector_endpoint: Uri,
     headers: HashMap<HeaderName, HeaderValue>,
     _timeout: Duration,
+    #[allow(dead_code)]
+    // <allow dead> would be removed once we support set_resource for metrics and traces.
+    resource: opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema,
 }
 
 impl OtlpHttpClient {
@@ -270,6 +274,7 @@
             collector_endpoint,
             headers,
             _timeout: timeout,
+            resource: ResourceAttributesWithSchema::default(),
         }
     }
 }

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

@@ -1,6 +1,5 @@
-use core::fmt;
-
 use async_trait::async_trait;
+use core::fmt;
 use opentelemetry::logs::{LogError, LogResult};
 use opentelemetry_proto::tonic::collector::logs::v1::{
     logs_service_client::LogsServiceClient, ExportLogsServiceRequest,
@@ -12,6 +11,9 @@
 
 pub(crate) struct TonicLogsClient {
     inner: Option<ClientInner>,
+    #[allow(dead_code)]
+    // <allow dead> would be removed once we support set_resource for metrics and traces.
+    resource: opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema,
 }
 
 struct ClientInner {
@@ -43,6 +45,7 @@
                 client,
                 interceptor,
             }),
+            resource: Default::default(),
         }
     }
 }
@@ -62,13 +65,19 @@
             None => return Err(LogError::Other("exporter is already shut down".into())),
         };
 
+        let resource_logs = {
+            batch
+                .into_iter()
+                .map(|log_data| (log_data, &self.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)?;
@@ -79,4 +88,8 @@
     fn shutdown(&mut self) {
         let _ = self.inner.take();
     }
+
+    fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {
+        self.resource = resource.into();
+    }
 }

opentelemetry-otlp/src/logs.rs

@@ -103,6 +103,10 @@
     async fn export(&mut self, batch: Vec<LogData>) -> opentelemetry::logs::LogResult<()> {
         self.client.export(batch).await
     }
+
+    fn set_resource(&mut self, _resource: &opentelemetry_sdk::Resource) {
+        todo!("set_resource");
+    }
 }
 
 /// Recommended configuration for an OTLP exporter pipeline.

opentelemetry-proto/src/transform/common.rs

@@ -22,6 +22,23 @@
     use opentelemetry::{Array, Value};
     use std::borrow::Cow;
 
+    #[cfg(any(feature = "trace", feature = "logs"))]
+    #[derive(Debug, Default)]
+    pub struct ResourceAttributesWithSchema {
+        pub attributes: Attributes,
+        pub schema_url: Option<String>,
+    }
+
+    #[cfg(any(feature = "trace", feature = "logs"))]
+    impl From<&opentelemetry_sdk::Resource> for ResourceAttributesWithSchema {
+        fn from(resource: &opentelemetry_sdk::Resource) -> Self {
+            ResourceAttributesWithSchema {
+                attributes: resource_attributes(resource),
+                schema_url: resource.schema_url().map(ToString::to_string),
+            }
+        }
+    }
+
     #[cfg(any(feature = "trace", feature = "logs"))]
     use opentelemetry_sdk::Resource;
 
@@ -52,7 +69,7 @@
     }
 
     /// Wrapper type for Vec<`KeyValue`>
-    #[derive(Default)]
+    #[derive(Default, Debug)]
     pub struct Attributes(pub ::std::vec::Vec<crate::proto::tonic::common::v1::KeyValue>);
 
     impl From<Vec<opentelemetry::KeyValue>> for Attributes {

opentelemetry-proto/src/transform/logs.rs

@@ -7,7 +7,7 @@
             resource::v1::Resource,
             Attributes,
         },
-        transform::common::{to_nanos, tonic::resource_attributes},
+        transform::common::{to_nanos, tonic::ResourceAttributesWithSchema},
     };
     use opentelemetry::logs::{AnyValue as LogsAnyValue, Severity};
 
@@ -110,18 +110,26 @@
         }
     }
 
-    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::LogData,
+            &ResourceAttributesWithSchema,
+        )> for ResourceLogs
+    {
+        fn from(
+            data: (
+                opentelemetry_sdk::export::logs::LogData,
+                &ResourceAttributesWithSchema,
+            ),
+        ) -> Self {
+            let (log_data, resource) = data;
+
             ResourceLogs {
                 resource: Some(Resource {
-                    attributes: resource_attributes(&log_data.resource).0,
+                    attributes: resource.attributes.0.clone(),
                     dropped_attributes_count: 0,
                 }),
-                schema_url: log_data
-                    .resource
-                    .schema_url()
-                    .map(Into::into)
-                    .unwrap_or_default(),
+                schema_url: resource.schema_url.clone().unwrap(),
                 scope_logs: vec![ScopeLogs {
                     schema_url: log_data
                         .instrumentation

opentelemetry-sdk/CHANGELOG.md

@@ -12,6 +12,10 @@
 - **Breaking** [#1624](https://github.com/open-telemetry/opentelemetry-rust/pull/1624) Remove `OsResourceDetector` and
   `ProcessResourceDetector` resource detectors, use the
   [`opentelemetry-resource-detector`](https://crates.io/crates/opentelemetry-resource-detectors) instead.
+- [#1636](https://github.com/open-telemetry/opentelemetry-rust/pull/1636) [Logs SDK] Send resource attributes once to 
+  processor and exporter, and not for every event.
+    - Add `set_resource` method in LogProcessor an Exporter
+    - Propagate resource attributes to processor and exporter during LoggerProvider creation.
 
 ## v0.22.1
 

opentelemetry-sdk/benches/log.rs

@@ -20,6 +20,8 @@ impl LogExporter for VoidExporter {
     async fn export(&mut self, _batch: Vec<LogData>) -> LogResult<()> {
         LogResult::Ok(())
     }
+
+    fn set_resource(&mut self, _resource: &opentelemetry_sdk::Resource) {}
 }
 
 fn log_benchmark_group<F: Fn(&dyn Logger)>(c: &mut Criterion, name: &str, f: F) {

opentelemetry-sdk/src/export/logs/mod.rs

@@ -7,7 +7,7 @@ use opentelemetry::{
     logs::{LogError, LogRecord, LogResult},
     InstrumentationLibrary,
 };
-use std::{borrow::Cow, fmt::Debug};
+use std::fmt::Debug;
 
 /// `LogExporter` defines the interface that log exporters should implement.
 #[async_trait]
@@ -21,17 +21,16 @@ pub trait LogExporter: Send + Sync + Debug {
     fn event_enabled(&self, _level: Severity, _target: &str, _name: &str) -> bool {
         true
     }
+    /// Set the resource for the exporter.
+    fn set_resource(&mut self, resource: &Resource);
 }
 
-/// `LogData` associates a [`LogRecord`] with a [`Resource`] and
-/// [`InstrumentationLibrary`].
+/// `LogData` represents a single log event without resource context.
 #[derive(Clone, Debug)]
 pub struct LogData {
     /// Log record
     pub record: LogRecord,
-    /// Resource for the emitter who produced this `LogData`.
-    pub resource: Cow<'static, Resource>,
-    /// Instrumentation details for the emitter who produced this `LogData`.
+    /// Instrumentation details for the emitter who produced this `LogEvent`.
     pub instrumentation: InstrumentationLibrary,
 }
 

opentelemetry-sdk/src/logs/log_emitter.rs

@@ -156,6 +156,10 @@ impl Builder {
 
     /// Create a new provider from this configuration.
     pub fn build(self) -> LoggerProvider {
+        // invoke set_resource on all the processors
+        for processor in &self.processors {
+            processor.set_resource(&self.config.resource);
+        }
         LoggerProvider {
             inner: Arc::new(LoggerProviderInner {
                 processors: self.processors,
@@ -200,7 +204,6 @@ impl opentelemetry::logs::Logger for Logger {
     /// Emit a `LogRecord`.
     fn emit(&self, record: LogRecord) {
         let provider = self.provider();
-        let config = provider.config();
         let processors = provider.log_processors();
         let trace_context = Context::map_current(|cx| {
             cx.has_active_span()
@@ -213,7 +216,6 @@ impl opentelemetry::logs::Logger for Logger {
             }
             let data = LogData {
                 record,
-                resource: config.resource.clone(),
                 instrumentation: self.instrumentation_library().clone(),
             };
             p.emit(data);
@@ -363,5 +365,9 @@ mod tests {
         fn event_enabled(&self, _level: Severity, _target: &str, _name: &str) -> bool {
             true
         }
+
+        fn set_resource(&self, _resource: &crate::Resource) {
+            // nothing to do.
+        }
     }
 }