Fix userevents exporter (#191)

Author: cijothomas

.cspell.json

@@ -29,18 +29,22 @@
         "appender",
         "appenders",
         "Bhasin",
+        "chrono",
         "Cijo",
         "codecov",
         "deque",
         "Dirkjan",
         "errno",
+        "eventheader",
+        "ftrace",
         "hasher",
         "isahc",
         "Isobel",
         "jaegertracing",
         "Kühle",
         "Kumar",
         "Lalit",
+        "microbench",
         "msrv",
         "myprovider",
         "Ochtman",
@@ -54,6 +58,7 @@
         "rustc",
         "Tescher",
         "tracepoint",
+        "Tracepoints",
         "Zhongyang",
         "zipkin"
     ],

opentelemetry-user-events-logs/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "opentelemetry-user-events-logs"
-description = "OpenTelemetry-Rust exporter to userevents"
+description = "OpenTelemetry-Rust exporter to user_events"
 version = "0.10.0"
 edition = "2021"
 homepage = "https://github.com/open-telemetry/opentelemetry-rust-contrib/tree/main/opentelemetry-user-events-logs"
@@ -17,14 +17,14 @@ opentelemetry = { workspace = true, features = ["logs"] }
 opentelemetry_sdk = { workspace = true, features = ["logs"] }
 chrono = { version = "0.4", default-features = false, features = ["std"] }
 tracing = { version = "0.1", optional = true }
+futures-executor = "0.3"
 
 [dev-dependencies]
 opentelemetry-appender-tracing = { workspace = true }
 tracing = { version = "0.1", default-features = false, features = ["std"] }
 tracing-core = "0.1.31"
 tracing-subscriber = { version = "0.3.0", default-features = false, features = ["env-filter", "fmt", "registry", "std"] }
-microbench = "0.5"
-ctrlc = "3.4" 
+ctrlc = "3.4"
 
 [features]
 spec_unstable_logs_enabled = ["opentelemetry/spec_unstable_logs_enabled", "opentelemetry_sdk/spec_unstable_logs_enabled", "opentelemetry-appender-tracing/spec_unstable_logs_enabled"]

opentelemetry-user-events-logs/README.md

@@ -16,13 +16,15 @@ Windows. It builds on top of the Linux Tracepoints, and so allows user processes
 to create events and trace data that can be viewed via existing tools like
 ftrace and perf.
 
-This kernel feature is supported started in Linux kernel 5.18 onwards. The feature enables
- - A faster path for tracing from user mode application utilizing kernel mode memory address space. 
- - User processes can now export telemetry events only when it is useful i.e, when the registered set of tracepoint events are enabled.
+This kernel feature is supported started in Linux kernel 6.4 onwards. The feature enables
 
- This user_events exporter enables applications to use OpenTelemetry API to capture the telemetry events, and write to user_events subsystem. From user_events, the events can be
-  - Captured by the agents running locally, and listening for specific events withing user_events subsystem.
-  - Or real-time monitoring using local Linux tool like [perf](https://perf.wiki.kernel.org/index.php/Main_Page) or ftrace.
+- A faster path for tracing from user mode application utilizing kernel mode memory address space.
+- User processes can now export telemetry events only when it is useful i.e, when the registered set of tracepoint events are enabled.
+
+This user_events exporter enables applications to use OpenTelemetry API to capture the telemetry events, and write to user_events subsystem. From user_events, the events can be
+
+- Captured by the agents running locally, and listening for specific events withing user_events subsystem.
+- Or real-time monitoring using local Linux tool like [perf](https://perf.wiki.kernel.org/index.php/Main_Page) or ftrace.
 
 [![Crates.io: opentelemetry-user-events-logs](https://img.shields.io/crates/v/opentelemetry-user-events-logs.svg)](https://crates.io/crates/opentelemetry-user-events-logs)
 [![Documentation](https://docs.rs/opentelemetry-user-events-logs/badge.svg)](https://docs.rs/opentelemetry-user-events-logs)

opentelemetry-user-events-logs/src/logs/exporter.rs

@@ -383,17 +383,11 @@ impl Debug for UserEventsExporter {
 }
 
 impl opentelemetry_sdk::logs::LogExporter for UserEventsExporter {
-    #[allow(clippy::manual_async_fn)]
-    fn export(
-        &self,
-        batch: opentelemetry_sdk::logs::LogBatch<'_>,
-    ) -> impl std::future::Future<Output = OTelSdkResult> + Send {
-        async move {
-            for (record, instrumentation) in batch.iter() {
-                let _ = self.export_log_data(record, instrumentation);
-            }
-            Ok(())
+    async fn export(&self, batch: opentelemetry_sdk::logs::LogBatch<'_>) -> OTelSdkResult {
+        for (record, instrumentation) in batch.iter() {
+            let _ = self.export_log_data(record, instrumentation);
         }
+        Ok(())
     }
 
     #[cfg(feature = "spec_unstable_logs_enabled")]

opentelemetry-user-events-logs/src/logs/reentrant_logprocessor.rs

@@ -1,47 +1,40 @@
 use std::fmt::Debug;
 
-use opentelemetry_sdk::error::OTelSdkResult;
-
-#[cfg(feature = "spec_unstable_logs_enabled")]
+use opentelemetry::InstrumentationScope;
 use opentelemetry_sdk::logs::LogExporter;
-
-use crate::logs::exporter::*;
+use opentelemetry_sdk::{
+    error::OTelSdkResult,
+    logs::{LogBatch, SdkLogRecord},
+};
 
 /// This export processor exports without synchronization.
 /// This is currently only used in users_event exporter, where we know
 /// that the underlying exporter is safe under concurrent calls
 
 #[derive(Debug)]
-pub struct ReentrantLogProcessor {
-    event_exporter: UserEventsExporter,
+pub struct ReentrantLogProcessor<T: LogExporter> {
+    exporter: T,
 }
 
-impl ReentrantLogProcessor {
+impl<T: LogExporter> ReentrantLogProcessor<T> {
     /// constructor that accepts an exporter instance
-    pub fn new(exporter: UserEventsExporter) -> Self {
-        ReentrantLogProcessor {
-            event_exporter: exporter,
-        }
+    pub fn new(exporter: T) -> Self {
+        ReentrantLogProcessor { exporter }
     }
 }
 
-impl opentelemetry_sdk::logs::LogProcessor for ReentrantLogProcessor {
-    fn emit(
-        &self,
-        record: &mut opentelemetry_sdk::logs::SdkLogRecord,
-        instrumentation: &opentelemetry::InstrumentationScope,
-    ) {
-        _ = self.event_exporter.export_log_data(record, instrumentation);
+impl<T: LogExporter> opentelemetry_sdk::logs::LogProcessor for ReentrantLogProcessor<T> {
+    fn emit(&self, record: &mut SdkLogRecord, scope: &InstrumentationScope) {
+        let log_tuple = &[(record as &SdkLogRecord, scope)];
+        let _ = futures_executor::block_on(self.exporter.export(LogBatch::new(log_tuple)));
     }
 
-    // This is a no-op as this processor doesn't keep anything
-    // in memory to be flushed out.
+    // Nothing to flush
     fn force_flush(&self) -> OTelSdkResult {
         Ok(())
     }
 
-    // This is a no-op no special cleanup is required before
-    // shutdown.
+    // Nothing to shutdown
     fn shutdown(&self) -> OTelSdkResult {
         Ok(())
     }
@@ -53,6 +46,6 @@ impl opentelemetry_sdk::logs::LogProcessor for ReentrantLogProcessor {
         target: &str,
         name: &str,
     ) -> bool {
-        self.event_exporter.event_enabled(level, target, name)
+        self.exporter.event_enabled(level, target, name)
     }
 }

stress/Cargo.toml

@@ -32,7 +32,7 @@ opentelemetry_sdk = { workspace = true, features = ["logs", "spec_unstable_logs_
 opentelemetry-user-events-logs = { path = "../opentelemetry-user-events-logs", features = ["spec_unstable_logs_enabled"]}
 opentelemetry-etw-logs = { path = "../opentelemetry-etw-logs"}
 tracing = { version = "0.1", default-features = false, features = ["std"] }
-tracing-subscriber = { version = "0.3.0", default-features = false, features = ["registry", "std"] }
+tracing-subscriber = { version = "0.3.0", default-features = false, features = ["env-filter","registry", "std"] }
 
 [features]
 stats = ["sysinfo"]

stress/src/user_events.rs

@@ -26,7 +26,7 @@ use opentelemetry_appender_tracing::layer;
 use opentelemetry_sdk::logs::SdkLoggerProvider;
 use opentelemetry_user_events_logs::UserEventsLoggerProviderBuilderExt;
 use tracing::info;
-use tracing_subscriber::prelude::*;
+use tracing_subscriber::{prelude::*, EnvFilter};
 mod throughput;
 
 // Function to initialize the logger
@@ -49,7 +49,9 @@ fn log_event_task() {
 fn main() {
     // Initialize the logger
     let logger_provider = init_logger();
+    let filter_otel = EnvFilter::new("info").add_directive("opentelemetry=off".parse().unwrap());
     let layer = layer::OpenTelemetryTracingBridge::new(&logger_provider);
+    let layer = layer.with_filter(filter_otel);
     tracing_subscriber::registry().with(layer).init();
 
     // Use the provided stress test framework