Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

initial commit #30

Open
wants to merge 14 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions opentelemetry-appender-tracing/benches/logs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,8 @@ use criterion::{criterion_group, criterion_main, Criterion};
use opentelemetry::logs::LogResult;
use opentelemetry::KeyValue;
use opentelemetry_appender_tracing::layer as tracing_layer;
use opentelemetry_sdk::export::logs::{LogData, LogExporter};
use opentelemetry_sdk::logs::{LogProcessor, LoggerProvider};
use opentelemetry_sdk::export::logs::LogExporter;
use opentelemetry_sdk::logs::{LogData, LogProcessor, LoggerProvider};
use opentelemetry_sdk::Resource;
use pprof::criterion::{Output, PProfProfiler};
use tracing::error;
Expand All @@ -34,7 +34,13 @@ struct NoopExporter {

#[async_trait]
impl LogExporter for NoopExporter {
async fn export<'a>(&mut self, _: Vec<std::borrow::Cow<'a, LogData>>) -> LogResult<()> {
async fn export(
&mut self,
_: Vec<(
&opentelemetry_sdk::logs::LogRecord,
&opentelemetry::InstrumentationLibrary,
)>,
) -> LogResult<()> {
LogResult::Ok(())
}

Expand Down
14 changes: 5 additions & 9 deletions opentelemetry-otlp/src/exporter/http/logs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,15 @@ use std::sync::Arc;
use async_trait::async_trait;
use http::{header::CONTENT_TYPE, Method};
use opentelemetry::logs::{LogError, LogResult};
use opentelemetry_sdk::export::logs::{LogData, LogExporter};
use opentelemetry::InstrumentationLibrary;
use opentelemetry_sdk::export::logs::LogExporter;
use opentelemetry_sdk::logs::LogRecord;

use super::OtlpHttpClient;

#[async_trait]
impl LogExporter for OtlpHttpClient {
async fn export<'a>(&mut self, batch: Vec<std::borrow::Cow<'a, LogData>>) -> LogResult<()> {
async fn export(&mut self, batch: Vec<(&LogRecord, &InstrumentationLibrary)>) -> LogResult<()> {
let client = self
.client
.lock()
Expand All @@ -19,13 +21,7 @@ impl LogExporter for OtlpHttpClient {
_ => Err(LogError::Other("exporter is already shut down".into())),
})?;

//TODO: avoid cloning here.
let owned_batch = batch
.into_iter()
.map(|cow_log_data| cow_log_data.into_owned()) // Converts Cow to owned LogData
.collect::<Vec<LogData>>();

let (body, content_type) = { self.build_logs_export_body(owned_batch)? };
let (body, content_type) = { self.build_logs_export_body(batch)? };
let mut request = http::Request::builder()
.method(Method::POST)
.uri(&self.collector_endpoint)
Expand Down
8 changes: 5 additions & 3 deletions opentelemetry-otlp/src/exporter/http/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7,16 +7,18 @@ use crate::{
OTEL_EXPORTER_OTLP_TIMEOUT,
};
use http::{HeaderName, HeaderValue, Uri};
#[cfg(feature = "logs")]
use opentelemetry::InstrumentationLibrary;
use opentelemetry_http::HttpClient;
use opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema;
#[cfg(feature = "logs")]
use opentelemetry_proto::transform::logs::tonic::group_logs_by_resource_and_scope;
#[cfg(feature = "trace")]
use opentelemetry_proto::transform::trace::tonic::group_spans_by_resource_and_scope;
#[cfg(feature = "logs")]
use opentelemetry_sdk::export::logs::LogData;
#[cfg(feature = "trace")]
use opentelemetry_sdk::export::trace::SpanData;
#[cfg(feature = "logs")]
use opentelemetry_sdk::logs::LogRecord;
#[cfg(feature = "metrics")]
use opentelemetry_sdk::metrics::data::ResourceMetrics;
use prost::Message;
Expand Down Expand Up @@ -328,7 +330,7 @@ impl OtlpHttpClient {
#[cfg(feature = "logs")]
fn build_logs_export_body(
&self,
logs: Vec<LogData>,
logs: Vec<(&LogRecord, &InstrumentationLibrary)>,
) -> opentelemetry::logs::LogResult<(Vec<u8>, &'static str)> {
use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
let resource_logs = group_logs_by_resource_and_scope(logs, &self.resource);
Expand Down
15 changes: 6 additions & 9 deletions opentelemetry-otlp/src/exporter/tonic/logs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,13 +4,16 @@ use opentelemetry::logs::{LogError, LogResult};
use opentelemetry_proto::tonic::collector::logs::v1::{
logs_service_client::LogsServiceClient, ExportLogsServiceRequest,
};
use opentelemetry_sdk::export::logs::{LogData, LogExporter};
use opentelemetry_sdk::export::logs::LogExporter;
use tonic::{codegen::CompressionEncoding, service::Interceptor, transport::Channel, Request};

use opentelemetry_proto::transform::logs::tonic::group_logs_by_resource_and_scope;

use super::BoxInterceptor;

use opentelemetry::InstrumentationLibrary;
use opentelemetry_sdk::logs::LogRecord;

pub(crate) struct TonicLogsClient {
inner: Option<ClientInner>,
#[allow(dead_code)]
Expand Down Expand Up @@ -54,7 +57,7 @@ impl TonicLogsClient {

#[async_trait]
impl LogExporter for TonicLogsClient {
async fn export<'a>(&mut self, batch: Vec<std::borrow::Cow<'a, LogData>>) -> LogResult<()> {
async fn export(&mut self, batch: Vec<(&LogRecord, &InstrumentationLibrary)>) -> LogResult<()> {
let (mut client, metadata, extensions) = match &mut self.inner {
Some(inner) => {
let (m, e, _) = inner
Expand All @@ -67,13 +70,7 @@ impl LogExporter for TonicLogsClient {
None => return Err(LogError::Other("exporter is already shut down".into())),
};

//TODO: avoid cloning here.
let owned_batch = batch
.into_iter()
.map(|cow_log_data| cow_log_data.into_owned()) // Converts Cow to owned LogData
.collect::<Vec<LogData>>();

let resource_logs = group_logs_by_resource_and_scope(owned_batch, &self.resource);
let resource_logs = group_logs_by_resource_and_scope(batch, &self.resource);

client
.export(Request::from_parts(
Expand Down
8 changes: 5 additions & 3 deletions opentelemetry-otlp/src/logs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,9 @@ use std::fmt::Debug;

use opentelemetry::logs::LogError;

use opentelemetry_sdk::{export::logs::LogData, runtime::RuntimeChannel, Resource};
use opentelemetry::InstrumentationLibrary;
use opentelemetry_sdk::logs::LogRecord;
use opentelemetry_sdk::{runtime::RuntimeChannel, Resource};

/// Compression algorithm to use, defaults to none.
pub const OTEL_EXPORTER_OTLP_LOGS_COMPRESSION: &str = "OTEL_EXPORTER_OTLP_LOGS_COMPRESSION";
Expand Down Expand Up @@ -98,9 +100,9 @@ impl LogExporter {

#[async_trait]
impl opentelemetry_sdk::export::logs::LogExporter for LogExporter {
async fn export<'a>(
async fn export(
&mut self,
batch: Vec<std::borrow::Cow<'a, LogData>>,
batch: Vec<(&LogRecord, &InstrumentationLibrary)>,
) -> opentelemetry::logs::LogResult<()> {
self.client.export(batch).await
}
Expand Down
82 changes: 47 additions & 35 deletions opentelemetry-proto/src/transform/logs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -54,8 +54,8 @@ pub mod tonic {
}
}

impl From<opentelemetry_sdk::logs::LogRecord> for LogRecord {
fn from(log_record: opentelemetry_sdk::logs::LogRecord) -> Self {
impl From<&opentelemetry_sdk::logs::LogRecord> for LogRecord {
fn from(log_record: &opentelemetry_sdk::logs::LogRecord) -> Self {
let trace_context = log_record.trace_context.as_ref();
let severity_number = match log_record.severity_number {
Some(Severity::Trace) => SeverityNumber::Trace,
Expand Down Expand Up @@ -118,7 +118,7 @@ pub mod tonic {
},
severity_number: severity_number.into(),
severity_text: log_record.severity_text.map(Into::into).unwrap_or_default(),
body: log_record.body.map(Into::into),
body: log_record.body.clone().map(Into::into),
dropped_attributes_count: 0,
flags: trace_context
.map(|ctx| {
Expand All @@ -139,17 +139,23 @@ pub mod tonic {

impl
From<(
opentelemetry_sdk::export::logs::LogData,
(
&opentelemetry_sdk::logs::LogRecord,
&opentelemetry::InstrumentationLibrary,
),
&ResourceAttributesWithSchema,
)> for ResourceLogs
{
fn from(
data: (
opentelemetry_sdk::export::logs::LogData,
(
&opentelemetry_sdk::logs::LogRecord,
&opentelemetry::InstrumentationLibrary,
),
&ResourceAttributesWithSchema,
),
) -> Self {
let (log_data, resource) = data;
let ((log_record, instrumentation), resource) = data;

ResourceLogs {
resource: Some(Resource {
Expand All @@ -158,37 +164,44 @@ pub mod tonic {
}),
schema_url: resource.schema_url.clone().unwrap_or_default(),
scope_logs: vec![ScopeLogs {
schema_url: log_data
.instrumentation
schema_url: instrumentation
.schema_url
.clone()
.map(Into::into)
.unwrap_or_default(),
scope: Some((log_data.instrumentation, log_data.record.target.clone()).into()),
log_records: vec![log_data.record.into()],
scope: Some((instrumentation, log_record.target.clone()).into()),
log_records: vec![log_record.into()],
}],
}
}
}

pub fn group_logs_by_resource_and_scope(
logs: Vec<opentelemetry_sdk::export::logs::LogData>,
logs: Vec<(
&opentelemetry_sdk::logs::LogRecord,
&opentelemetry::InstrumentationLibrary,
)>,
resource: &ResourceAttributesWithSchema,
) -> Vec<ResourceLogs> {
// Group logs by target or instrumentation name
let scope_map = logs.iter().fold(
HashMap::new(),
|mut scope_map: HashMap<
Cow<'static, str>,
Vec<&opentelemetry_sdk::export::logs::LogData>,
Vec<(
&opentelemetry_sdk::logs::LogRecord,
&opentelemetry::InstrumentationLibrary,
)>,
>,
log| {
let key = log
.record
(log_record, instrumentation)| {
let key = log_record
.target
.clone()
.unwrap_or_else(|| log.instrumentation.name.clone());
scope_map.entry(key).or_default().push(log);
.unwrap_or_else(|| Cow::Owned(instrumentation.name.clone().into_owned()));
scope_map
.entry(key)
.or_default()
.push((log_record, instrumentation));
scope_map
},
);
Expand All @@ -197,13 +210,13 @@ pub mod tonic {
.into_iter()
.map(|(key, log_data)| ScopeLogs {
scope: Some(InstrumentationScope::from((
&log_data.first().unwrap().instrumentation,
Some(key),
log_data.first().unwrap().1,
Some(key.into_owned().into()),
))),
schema_url: resource.schema_url.clone().unwrap_or_default(),
log_records: log_data
.into_iter()
.map(|log_data| log_data.record.clone().into())
.map(|(log_record, _)| log_record.into())
.collect(),
})
.collect();
Expand All @@ -223,30 +236,29 @@ pub mod tonic {
mod tests {
use crate::transform::common::tonic::ResourceAttributesWithSchema;
use opentelemetry::logs::LogRecord as _;
use opentelemetry_sdk::export::logs::LogData;
use opentelemetry::InstrumentationLibrary;
use opentelemetry_sdk::{logs::LogRecord, Resource};
use std::time::SystemTime;

fn create_test_log_data(instrumentation_name: &str, _message: &str) -> LogData {
fn create_test_log_data(
instrumentation_name: &str,
_message: &str,
) -> (LogRecord, InstrumentationLibrary) {
let mut logrecord = LogRecord::default();
logrecord.set_timestamp(SystemTime::now());
logrecord.set_observed_timestamp(SystemTime::now());
LogData {
instrumentation: opentelemetry_sdk::InstrumentationLibrary::builder(
instrumentation_name.to_string(),
)
.build(),
record: logrecord,
}
let instrumentation =
InstrumentationLibrary::builder(instrumentation_name.to_string()).build();
(logrecord, instrumentation)
}

#[test]
fn test_group_logs_by_resource_and_scope_single_scope() {
let resource = Resource::default();
let log1 = create_test_log_data("test-lib", "Log 1");
let log2 = create_test_log_data("test-lib", "Log 2");
let log_data1 = create_test_log_data("test-lib", "Log 1");
let log_data2 = create_test_log_data("test-lib", "Log 2");

let logs = vec![log1, log2];
let logs = vec![(&log_data1.0, &log_data1.1), (&log_data2.0, &log_data2.1)];
let resource: ResourceAttributesWithSchema = (&resource).into(); // Convert Resource to ResourceAttributesWithSchema

let grouped_logs =
Expand All @@ -263,10 +275,10 @@ mod tests {
#[test]
fn test_group_logs_by_resource_and_scope_multiple_scopes() {
let resource = Resource::default();
let log1 = create_test_log_data("lib1", "Log 1");
let log2 = create_test_log_data("lib2", "Log 2");
let log_data1 = create_test_log_data("lib1", "Log 1");
let log_data2 = create_test_log_data("lib2", "Log 2");

let logs = vec![log1, log2];
let logs = vec![(&log_data1.0, &log_data1.1), (&log_data2.0, &log_data2.1)];
let resource: ResourceAttributesWithSchema = (&resource).into(); // Convert Resource to ResourceAttributesWithSchema
let grouped_logs =
crate::transform::logs::tonic::group_logs_by_resource_and_scope(logs, &resource);
Expand Down
38 changes: 38 additions & 0 deletions opentelemetry-sdk/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,44 @@
[#2021](https://github.com/open-telemetry/opentelemetry-rust/pull/2021)
- Provide default implementation for `event_enabled` method in `LogProcessor`
trait that returns `true` always.
- **Breaking** [#2035](https://github.com/open-telemetry/opentelemetry-rust/pull/2035)
- The Exporter::export() interface is modified as below:
Previous Signature:
```rust
async fn export<'a>(&mut self, batch: Vec<Cow<'a, LogData>>) -> LogResult<()>;
```

Updated Signature:
```rust
async fn export(&mut self, batch: Vec<(&LogRecord, &InstrumentationLibrary)>) -> LogResult<()>;
```
This change simplifies the processing required by exporters. Exporters no longer need to determine if the LogData is borrowed or owned, as they now work directly with references. As a result, exporters must explicitly create a copy of LogRecord and/or InstrumentationLibrary when needed, as the new interface only provides references to these structures.

- The LogData structure is NO longer the part of the export interface. So it has been moved from `opentelemetry_sdk::export::logs` to `opentelemetry_sdk::logs` namespace. The custom implementations of `LogProcessor` need to update the imports accordindgly.

- The LogData structure has been changed as below:
Previous Signature
```rust
#[derive(Clone, Debug)]
pub struct LogData {
/// Log record
pub record: LogRecord,
/// Instrumentation details for the emitter who produced this `LogEvent`.
pub instrumentation: InstrumentationLibrary,
```

Updated Signature:
```rust
#[derive(Clone, Debug)]
pub struct LogData<'a> {
/// Log record, which can be borrowed or owned.
pub record: Cow<'a, LogRecord>,
/// Instrumentation details for the emitter who produced this `LogEvent`.
pub instrumentation: Cow<'a, InstrumentationLibrary>,
}
```
The custom implementation of `LogProcessor` need to accordingly modify the handling of LogData
received through LogProcessor::emit() interface.

## v0.24.1

Expand Down
6 changes: 3 additions & 3 deletions opentelemetry-sdk/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -61,15 +61,15 @@ name = "span_builder"
harness = false

[[bench]]
name = "metric_counter"
name = "metrics_counter"
harness = false

[[bench]]
name = "metric_gauge"
name = "metrics_gauge"
harness = false

[[bench]]
name = "metric_histogram"
name = "metrics_histogram"
harness = false

[[bench]]
Expand Down
Loading
Loading