Skip to content

Commit

Permalink
feature: refactor and extend log_exporter benchmark
Browse files Browse the repository at this point in the history 
This commit refactors and extends the `log_exporter` benchmark.
1. Rename to be more precise: `FooWithFuture` becomes `FooAsyncTrait`, `FooWithoutFuture` becomes `FooSync`
2. Add variant using native Rust support for async traits
3. Share workload code between implementations
4. Use blackbox hint for implementations under test
5. Expand module documentation
6. Move each impl into its own submodule
  • Loading branch information
@demurgos
demurgos committed Sep 24, 2024
1 parent c8136d9 commit 21785fd
Showing 1 changed file with 184 additions and 106 deletions.
290 changes: 184 additions & 106 deletions opentelemetry-sdk/benches/log_exporter.rs
View file
Original file line number Diff line number Diff line change
@@ -1,22 +1,30 @@
/*
The benchmark results:
criterion = "0.5.1"
OS: Ubuntu 22.04.3 LTS (5.15.146.1-microsoft-standard-WSL2)
Hardware: AMD EPYC 7763 64-Core Processor - 2.44 GHz, 16vCPUs,
RAM: 64.0 GB
| Test | Average time|
|--------------------------------|-------------|
| LogExporterWithFuture | 111 ns |
| LogExporterWithoutFuture | 92 ns |
*/
//! Benchmark different definition styles for the `LogExporter` trait.
//!
//! Run this benchmark with the following command:
//! ```sh
//! cargo bench --bench log_exporter
//! ```
//!
//! Example benchmark results:
//! - date: 2024-09-24
//! - OS: Linux 6.10.10 x86_64 (Arch Linux)
//! - CPU: AMD Ryzen 9 7950X
//! - RAM: 64 GiB
//! ```text
//! LogExporterAsyncTrait time: [60.918 ns 61.146 ns 61.404 ns]
//! LogExporterAsyncNative time: [51.846 ns 51.905 ns 51.979 ns]
//! LogExporterSync time: [50.197 ns 50.332 ns 50.472 ns]
//! ```
use std::sync::Mutex;
use std::time::SystemTime;

use async_trait::async_trait;
use criterion::{criterion_group, criterion_main, Criterion};

use opentelemetry::logs::{LogRecord as _, LogResult, Logger as _, LoggerProvider as _, Severity};
use opentelemetry::logs::{
LogRecord as _, LogResult, Logger, LoggerProvider as _, Severity,
};

use opentelemetry::InstrumentationLibrary;
use opentelemetry_sdk::export::logs::LogBatch;
Expand All @@ -25,147 +33,217 @@ use opentelemetry_sdk::logs::LogRecord;
use opentelemetry_sdk::logs::LoggerProvider;
use pprof::criterion::{Output, PProfProfiler};
use std::fmt::Debug;
use std::hint::black_box;

// Run this benchmark with:
// cargo bench --bench log_exporter
#[async_trait]
pub trait LogExporterWithFuture: Send + Sync + Debug {
async fn export(&mut self, batch: LogBatch<'_>);
}
mod style_async_trait {
use super::*;

pub trait LogExporterWithoutFuture: Send + Sync + Debug {
fn export(&mut self, batch: LogBatch<'_>);
}
/// Async style using the `async_trait` crate
#[async_trait]
pub trait LogExporterAsyncTrait: Send + Sync + Debug {
async fn export(&mut self, batch: LogBatch<'_>);
}

#[derive(Debug)]
struct NoOpExporterWithFuture {}
#[derive(Debug)]
pub struct NoOpExporterAsyncTrait;

#[async_trait]
impl LogExporterWithFuture for NoOpExporterWithFuture {
async fn export(&mut self, _batch: LogBatch<'_>) {}
}
#[async_trait]
impl LogExporterAsyncTrait for NoOpExporterAsyncTrait {
async fn export(&mut self, batch: LogBatch<'_>) {
black_box(batch);
}
}

#[derive(Debug)]
struct NoOpExporterWithoutFuture {}
impl LogExporterWithoutFuture for NoOpExporterWithoutFuture {
fn export(&mut self, _batch: LogBatch<'_>) {}
}
#[derive(Debug)]
pub struct ExportingProcessorAsyncTrait {
exporter: Mutex<NoOpExporterAsyncTrait>,
}

#[derive(Debug)]
struct ExportingProcessorWithFuture {
exporter: Mutex<NoOpExporterWithFuture>,
}
impl ExportingProcessorAsyncTrait {
pub fn new(exporter: NoOpExporterAsyncTrait) -> Self {
Self {
exporter: Mutex::new(exporter),
}
}
}

impl LogProcessor for ExportingProcessorAsyncTrait {
fn emit(&self, record: &mut LogRecord, library: &InstrumentationLibrary) {
let mut exporter = self.exporter.lock().expect("lock error");
let logs = [(record as &LogRecord, library)];
futures_executor::block_on(exporter.export(LogBatch::new(&logs)));
}

impl ExportingProcessorWithFuture {
fn new(exporter: NoOpExporterWithFuture) -> Self {
ExportingProcessorWithFuture {
exporter: Mutex::new(exporter),
fn force_flush(&self) -> LogResult<()> {
Ok(())
}

fn shutdown(&self) -> LogResult<()> {
Ok(())
}
}
}

impl LogProcessor for ExportingProcessorWithFuture {
fn emit(&self, record: &mut LogRecord, library: &InstrumentationLibrary) {
let mut exporter = self.exporter.lock().expect("lock error");
let logs = [(record as &LogRecord, library)];
futures_executor::block_on(exporter.export(LogBatch::new(&logs)));
mod style_async_native {
use super::*;

/// Async style using Rustc support for async traits
pub trait LogExporterAsyncNative: Send + Sync + Debug {
fn export(&mut self, batch: LogBatch<'_>) -> impl std::future::Future<Output = ()> + Send;
}

fn force_flush(&self) -> LogResult<()> {
Ok(())
#[derive(Debug)]
pub struct NoOpExporterAsyncNative;

impl LogExporterAsyncNative for NoOpExporterAsyncNative {
#[inline]
async fn export(&mut self, batch: LogBatch<'_>) {
black_box(batch);
}
}

fn shutdown(&self) -> LogResult<()> {
Ok(())
#[derive(Debug)]
pub struct ExportingProcessorAsyncNative {
exporter: Mutex<NoOpExporterAsyncNative>,
}
}

#[derive(Debug)]
struct ExportingProcessorWithoutFuture {
exporter: Mutex<NoOpExporterWithoutFuture>,
}
impl ExportingProcessorAsyncNative {
pub fn new(exporter: NoOpExporterAsyncNative) -> Self {
Self {
exporter: Mutex::new(exporter),
}
}
}

impl LogProcessor for ExportingProcessorAsyncNative {
#[inline]
fn emit(&self, record: &mut LogRecord, library: &InstrumentationLibrary) {
let mut exporter = self.exporter.lock().expect("lock error");
let logs = [(record as &LogRecord, library)];
futures_executor::block_on(exporter.export(LogBatch::new(&logs)));
}

fn force_flush(&self) -> LogResult<()> {
Ok(())
}

impl ExportingProcessorWithoutFuture {
fn new(exporter: NoOpExporterWithoutFuture) -> Self {
ExportingProcessorWithoutFuture {
exporter: Mutex::new(exporter),
fn shutdown(&self) -> LogResult<()> {
Ok(())
}
}
}

impl LogProcessor for ExportingProcessorWithoutFuture {
fn emit(&self, record: &mut LogRecord, library: &InstrumentationLibrary) {
let logs = [(record as &LogRecord, library)];
self.exporter
.lock()
.expect("lock error")
.export(LogBatch::new(&logs));
mod style_sync {
use super::*;

/// Sync style
pub trait LogExporterSync: Send + Sync + Debug {
fn export(&mut self, batch: LogBatch<'_>);
}

fn force_flush(&self) -> LogResult<()> {
Ok(())
#[derive(Debug)]
pub struct NoOpExporterSync;

impl LogExporterSync for NoOpExporterSync {
fn export(&mut self, batch: LogBatch<'_>) {
black_box(batch);
}
}

fn shutdown(&self) -> LogResult<()> {
Ok(())
#[derive(Debug)]
pub struct ExportingProcessorSync {
exporter: Mutex<NoOpExporterSync>,
}

impl ExportingProcessorSync {
pub fn new(exporter: NoOpExporterSync) -> Self {
Self {
exporter: Mutex::new(exporter),
}
}
}

impl LogProcessor for ExportingProcessorSync {
fn emit(&self, record: &mut LogRecord, library: &InstrumentationLibrary) {
let mut exporter = self.exporter.lock().expect("lock error");
let logs = [(record as &LogRecord, library)];
exporter.export(LogBatch::new(&logs));
}

fn force_flush(&self) -> LogResult<()> {
Ok(())
}

fn shutdown(&self) -> LogResult<()> {
Ok(())
}
}
}

fn criterion_benchmark(c: &mut Criterion) {
exporter_with_future(c);
exporter_without_future(c);
bench_async_trait(c);
bench_async_native(c);
bench_sync(c);
}

fn exporter_with_future(c: &mut Criterion) {
fn bench_async_trait(c: &mut Criterion) {
let provider = LoggerProvider::builder()
.with_log_processor(ExportingProcessorWithFuture::new(NoOpExporterWithFuture {}))
.with_log_processor(style_async_trait::ExportingProcessorAsyncTrait::new(
style_async_trait::NoOpExporterAsyncTrait,
))
.build();
let logger = provider.logger("benchmark");

c.bench_function("LogExporterAsyncTrait", |b| {
b.iter(|| workload(&logger));
});
}

fn bench_async_native(c: &mut Criterion) {
let provider = LoggerProvider::builder()
.with_log_processor(style_async_native::ExportingProcessorAsyncNative::new(
style_async_native::NoOpExporterAsyncNative,
))
.build();
let logger = provider.logger("benchmark");

c.bench_function("LogExporterWithFuture", |b| {
b.iter(|| {
let mut log_record = logger.create_log_record();
let now = SystemTime::now();
log_record.set_observed_timestamp(now);
log_record.set_target("my-target".to_string());
log_record.set_event_name("CheckoutFailed");
log_record.set_severity_number(Severity::Warn);
log_record.set_severity_text("WARN");
log_record.add_attribute("book_id", "12345");
log_record.add_attribute("book_title", "Rust Programming Adventures");
log_record.add_attribute("message", "Unable to process checkout.");

logger.emit(log_record);
});
c.bench_function("LogExporterAsyncNative", |b| {
b.iter(|| workload(&logger));
});
}

fn exporter_without_future(c: &mut Criterion) {
fn bench_sync(c: &mut Criterion) {
let provider = LoggerProvider::builder()
.with_log_processor(ExportingProcessorWithoutFuture::new(
NoOpExporterWithoutFuture {},
.with_log_processor(style_sync::ExportingProcessorSync::new(
style_sync::NoOpExporterSync,
))
.build();
let logger = provider.logger("benchmark");

c.bench_function("LogExporterWithoutFuture", |b| {
b.iter(|| {
let mut log_record = logger.create_log_record();
let now = SystemTime::now();
log_record.set_observed_timestamp(now);
log_record.set_target("my-target".to_string());
log_record.set_event_name("CheckoutFailed");
log_record.set_severity_number(Severity::Warn);
log_record.set_severity_text("WARN");
log_record.add_attribute("book_id", "12345");
log_record.add_attribute("book_title", "Rust Programming Adventures");
log_record.add_attribute("message", "Unable to process checkout.");

logger.emit(log_record);
});
c.bench_function("LogExporterSync", |b| {
b.iter(|| workload(&logger));
});
}

/// Dummy workload to exercise a logger
fn workload<TyLogger>(logger: &TyLogger)
where
TyLogger: Logger,
{
let mut log_record = logger.create_log_record();
let now = SystemTime::now();
log_record.set_observed_timestamp(now);
log_record.set_target("my-target".to_string());
log_record.set_event_name("CheckoutFailed");
log_record.set_severity_number(Severity::Warn);
log_record.set_severity_text("WARN");
log_record.add_attribute("book_id", "12345");
log_record.add_attribute("book_title", "Rust Programming Adventures");
log_record.add_attribute("message", "Unable to process checkout.");

logger.emit(log_record);
}

#[cfg(not(target_os = "windows"))]
criterion_group! {
name = benches;
Expand Down

0 comments on commit 21785fd

Please sign in to comment.