Internal logging code path: Raw logger support

rust/otap-dataflow/Cargo.toml

@@ -50,6 +50,8 @@ otap-df-pdata-otlp-model = { path = "./crates/pdata/src/otlp/model"}
 otap-df-config = { path = "crates/config" }
 otap-df-controller = { path = "crates/controller" }
 otap-df-otap = { path = "crates/otap" }
+otap-df-pdata = { path = "crates/pdata" }
+otap-df-telemetry = { path = "crates/telemetry" }
 quiver = { package = "otap-df-quiver", path = "crates/quiver" }
 data_engine_expressions = { path = "../experimental/query_engine/expressions" }
 data_engine_kql_parser = { path = "../experimental/query_engine/kql-parser" }

rust/otap-dataflow/benchmarks/Cargo.toml

@@ -26,6 +26,9 @@ otap-df-engine = { path = "../crates/engine"}
 otap-df-telemetry = { path="../crates/telemetry"}
 otap-df-pdata = { path="../crates/pdata", features = ["bench"]}
 
+tracing.workspace = true
+tracing-subscriber = { workspace = true, features = ["registry"] }
+
 fluke-hpack.workspace = true
 futures-channel.workspace = true
 futures.workspace = true
@@ -87,3 +90,7 @@ harness = false
 [[bench]]
 name = "otap_logs_view"
 harness = false
+
+[[bench]]
+name = "self_tracing"
+harness = false

rust/otap-dataflow/benchmarks/benches/self_tracing/main.rs

@@ -0,0 +1,199 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Benchmarks for the compact log formatter.
+//!
+//! These benchmarks emit a single tracing event but perform N
+//! encoding or encoding-and-formatting operations inside the callback
+//!
+//! Benchmark names follow the pattern: `group/description/N_events`
+//!
+//! Example: `encode/3_attrs/1000_events` = 300 µs → 300 ns per event
+
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
+use tracing::{Event, Subscriber};
+use tracing_subscriber::layer::Layer;
+use tracing_subscriber::prelude::*;
+use tracing_subscriber::registry::LookupSpan;
+
+use otap_df_pdata::otlp::ProtoBuffer;
+use otap_df_telemetry::self_tracing::{
+    ConsoleWriter, DirectLogRecordEncoder, LogRecord, SavedCallsite,
+};
+
+#[cfg(not(windows))]
+use tikv_jemallocator::Jemalloc;
+
+#[cfg(not(windows))]
+#[global_allocator]
+static GLOBAL: Jemalloc = Jemalloc;
+
+/// The operation to perform on each event within the layer.
+#[derive(Clone, Copy)]
+enum BenchOp {
+    /// Encode the event into a LogRecord only.
+    Encode,
+    /// Encode once, then format N times.
+    Format,
+    /// Encode and format together N times.
+    EncodeAndFormat,
+    /// Encode to protobuf N times.
+    EncodeProto,
+}
+
+/// A layer that performs a configurable operation N times per event.
+struct BenchLayer {
+    iterations: usize,
+    op: BenchOp,
+}
+
+impl BenchLayer {
+    fn new(iterations: usize, op: BenchOp) -> Self {
+        Self { iterations, op }
+    }
+}
+
+impl<S> Layer<S> for BenchLayer
+where
+    S: Subscriber + for<'a> LookupSpan<'a>,
+{
+    fn on_event(&self, event: &Event<'_>, _ctx: tracing_subscriber::layer::Context<'_, S>) {
+        match self.op {
+            BenchOp::Encode => {
+                for _ in 0..self.iterations {
+                    let record = LogRecord::new(event);
+                    let _ = std::hint::black_box(record);
+                }
+            }
+            BenchOp::Format => {
+                // Encode once, format N times
+                let record = LogRecord::new(event);
+                let writer = ConsoleWriter::no_color();
+                let callsite = SavedCallsite::new(event.metadata());
+
+                for _ in 0..self.iterations {
+                    let line = writer.format_log_record(&record, &callsite);
+                    let _ = std::hint::black_box(line);
+                }
+            }
+            BenchOp::EncodeAndFormat => {
+                let writer = ConsoleWriter::no_color();
+
+                for _ in 0..self.iterations {
+                    let record = LogRecord::new(event);
+                    let callsite = SavedCallsite::new(event.metadata());
+                    let line = writer.format_log_record(&record, &callsite);
+                    let _ = std::hint::black_box(line);
+                }
+            }
+            BenchOp::EncodeProto => {
+                let mut buf = ProtoBuffer::new();
+                let mut encoder = DirectLogRecordEncoder::new(&mut buf);
+                let callsite = SavedCallsite::new(event.metadata());
+
+                for _ in 0..self.iterations {
+                    encoder.clear();
+                    let size = encoder.encode_log_record(LogRecord::new(event), &callsite);
+                    let _ = std::hint::black_box(size);
+                }
+            }
+        }
+    }
+}
+
+/// Macro to generate benchmark functions for different attribute counts.
+/// Each variant emits a consistent log statement for fair comparison.
+macro_rules! emit_log {
+    (0) => {
+        tracing::info!("benchmark message")
+    };
+    (3) => {
+        tracing::info!(
+            attr_str = "value",
+            attr_int = 42,
+            attr_bool = true,
+            "benchmark message"
+        )
+    };
+    (10) => {
+        tracing::info!(
+            attr_str1 = "string1",
+            attr_bool1 = true,
+            attr_str2 = "string2",
+            attr_float1 = 1.234,
+            attr_int1 = 42i64,
+            attr_str3 = "string3",
+            attr_bool2 = false,
+            attr_float2 = 5.678,
+            attr_int2 = 100u64,
+            attr_str4 = "string4",
+            "benchmark message"
+        )
+    };
+}
+
+/// Run a benchmark with the given layer, invoking the log emitter.
+fn run_bench<L, F>(b: &mut criterion::Bencher<'_>, layer: L, emit: F)
+where
+    L: Layer<tracing_subscriber::Registry> + Send + Sync + 'static,
+    F: Fn(),
+{
+    let subscriber = tracing_subscriber::registry().with(layer);
+    let dispatch = tracing::Dispatch::new(subscriber);
+
+    b.iter(|| {
+        tracing::dispatcher::with_default(&dispatch, &emit);
+        std::hint::black_box(());
+    });
+}
+
+/// Benchmark a specific operation across different iteration counts.
+fn bench_op(c: &mut Criterion, group_name: &str, op: BenchOp) {
+    let mut group = c.benchmark_group(group_name);
+
+    for &iterations in &[100, 1000] {
+        for &(attr_count, attr_label) in &[(0, "0_attrs"), (3, "3_attrs"), (10, "10_attrs")] {
+            let id = BenchmarkId::new(attr_label, format!("{}_events", iterations));
+
+            let _ = group.bench_with_input(id, &iterations, |b, &iters| {
+                let layer = BenchLayer::new(iters, op);
+                match attr_count {
+                    0 => run_bench(b, layer, || emit_log!(0)),
+                    3 => run_bench(b, layer, || emit_log!(3)),
+                    _ => run_bench(b, layer, || emit_log!(10)),
+                }
+            });
+        }
+    }
+
+    group.finish();
+}
+
+fn bench_encode(c: &mut Criterion) {
+    bench_op(c, "encode", BenchOp::Encode);
+}
+
+fn bench_format(c: &mut Criterion) {
+    bench_op(c, "format", BenchOp::Format);
+}
+
+fn bench_encode_and_format(c: &mut Criterion) {
+    bench_op(c, "encode_and_format", BenchOp::EncodeAndFormat);
+}
+
+fn bench_encode_proto(c: &mut Criterion) {
+    bench_op(c, "encode_proto", BenchOp::EncodeProto);
+}
+
+#[allow(missing_docs)]
+mod bench_entry {
+    use super::*;
+
+    criterion_group!(
+        name = benches;
+        config = Criterion::default();
+        targets = bench_encode, bench_format, bench_encode_and_format, bench_encode_proto
+    );
+}
+
+criterion_main!(bench_entry::benches);

rust/otap-dataflow/crates/pdata/src/otlp/common.rs

@@ -30,7 +30,7 @@ use std::fmt;
 use std::fmt::Write;
 use std::sync::LazyLock;
 
-pub(in crate::otlp) struct ResourceArrays<'a> {
+pub(crate) struct ResourceArrays<'a> {
     pub id: Option<&'a UInt16Array>,
     pub dropped_attributes_count: Option<&'a UInt32Array>,
     pub schema_url: Option<StringArrayAccessor<'a>>,
@@ -123,14 +123,14 @@ impl<'a> TryFrom<&'a RecordBatch> for ResourceArrays<'a> {
     }
 }
 
-pub(in crate::otlp) struct ScopeArrays<'a> {
+pub(crate) struct ScopeArrays<'a> {
     pub name: Option<StringArrayAccessor<'a>>,
     pub version: Option<StringArrayAccessor<'a>>,
     pub dropped_attributes_count: Option<&'a UInt32Array>,
     pub id: Option<&'a UInt16Array>,
 }
 
-pub static SCOPE_ARRAY_DATA_TYPE: LazyLock<DataType> = LazyLock::new(|| {
+static SCOPE_ARRAY_DATA_TYPE: LazyLock<DataType> = LazyLock::new(|| {
     DataType::Struct(Fields::from(vec![
         Field::new(
             consts::NAME,
@@ -491,11 +491,15 @@ macro_rules! proto_encode_len_delimited_unknown_size {
     }};
 }
 
-pub(crate) fn encode_len_placeholder(buf: &mut ProtoBuffer) {
+/// Write a 4-byte length placeholder for later patching.
+/// Do not use directly, use proto_encode_len_delimited_unknown_size.
+pub fn encode_len_placeholder(buf: &mut ProtoBuffer) {
     buf.buffer.extend_from_slice(&[0x80, 0x80, 0x80, 0x00]);
 }
 
-pub(crate) fn patch_len_placeholder(
+/// Patch a previously written length placeholder with the actual length.
+/// Do not use directly, use proto_encode_len_delimited_unknown_size.
+pub fn patch_len_placeholder(
     buf: &mut ProtoBuffer,
     num_bytes: usize,
     len: usize,

rust/otap-dataflow/crates/pdata/src/otlp/mod.rs

@@ -9,25 +9,26 @@ use crate::{error::Result, otap::OtapArrowRecords};
 use bytes::Bytes;
 use otap_df_config::SignalType;
 
-pub use common::ProtoBuffer;
+pub use common::{ProtoBuffer, encode_len_placeholder, patch_len_placeholder};
 pub use otap_df_pdata_otlp_macros::Message; // Required for derived code
 pub use otap_df_pdata_otlp_macros::qualified; // Required for derived code
 
 /// Common methods for OTLP/OTAP attributes.
 pub mod attributes;
 /// Common methods for batching.
 pub mod batching;
+/// Common utilities for protobuf encoding.
+pub mod common;
 /// Common methods for OTLP/OTAP logs.
 pub mod logs;
 /// Common methods for OTLP/OTAP metrics.
 pub mod metrics;
 /// Common methods for OTLP/OTAP traces.
 pub mod traces;
 
-mod common;
-
 #[cfg(test)]
 mod batching_tests;
+
 #[cfg(test)]
 mod tests;
 

rust/otap-dataflow/crates/pdata/src/views/otlp/bytes/common.rs

@@ -37,8 +37,10 @@ pub struct RawKeyValue<'a> {
 }
 
 impl<'a> RawKeyValue<'a> {
+    /// Create a new RawKeyValue parser from a byte slice containing a KeyValue message.
     #[inline]
-    fn new(buf: &'a [u8]) -> Self {
+    #[must_use]
+    pub fn new(buf: &'a [u8]) -> Self {
         Self {
             buf,
             pos: Cell::new(0),

rust/otap-dataflow/crates/pdata/src/views/otlp/bytes/logs.rs

@@ -153,6 +153,18 @@ pub struct RawLogRecord<'a> {
     bytes_parser: ProtoBytesParser<'a, LogFieldOffsets>,
 }
 
+impl<'a> RawLogRecord<'a> {
+    /// Create a new instance of `RawLogRecord`. This is exposed
+    /// specifically for interpreting internally generated log records
+    /// which encode body and attributes as OTLP bytes.
+    #[must_use]
+    pub fn new(buf: &'a [u8]) -> Self {
+        Self {
+            bytes_parser: ProtoBytesParser::new(buf),
+        }
+    }
+}
+
 /// Known field offsets within byte buffer for fields in ResourceLogs message
 pub struct LogFieldOffsets {
     scalar_fields: [Cell<Option<(NonZeroUsize, NonZeroUsize)>>; 13],
@@ -274,11 +286,7 @@ impl<'a> Iterator for LogRecordsIter<'a> {
     type Item = RawLogRecord<'a>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        let slice = self.byte_parser.next()?;
-
-        Some(RawLogRecord {
-            bytes_parser: ProtoBytesParser::new(slice),
-        })
+        Some(RawLogRecord::new(self.byte_parser.next()?))
     }
 }
 

rust/otap-dataflow/crates/telemetry/Cargo.toml

@@ -19,8 +19,12 @@ unchecked-index = []
 unchecked-arithmetic = []
 
 [dependencies]
-axum = { workspace = true }
+otap-df-pdata = { workspace = true }
 otap-df-config = { workspace = true }
+
+axum = { workspace = true }
+bytes = { workspace = true }
+chrono = { workspace = true }
 flume = { workspace = true }
 tokio = { workspace = true }
 tokio-util = { workspace = true }
@@ -29,6 +33,7 @@ thiserror = { workspace = true }
 slotmap = { workspace = true }
 parking_lot = { workspace = true }
 prometheus = { workspace = true }
+prost = { workspace = true }
 serde = { workspace = true }
 tonic = { workspace = true, optional = true }
 opentelemetry = { workspace = true }

rust/otap-dataflow/crates/telemetry/src/lib.rs

@@ -40,6 +40,7 @@ pub mod metrics;
 pub mod opentelemetry_client;
 pub mod registry;
 pub mod reporter;
+pub mod self_tracing;
 pub mod semconv;
 
 // Re-export _private module from internal_events for macro usage.