open-telemetry · lquerel · Jan 8, 2026 · Dec 31, 2025 · Dec 31, 2025 · Jan 4, 2026
@@ -1,4 +1,4 @@
-# OTAP Dataflow Library
+# OTAP Dataflow Engine
 
 [![build](https://github.com/open-telemetry/otel-arrow/actions/workflows/rust-ci.yml/badge.svg)](https://github.com/open-telemetry/otel-arrow/actions/workflows/rust-ci.yml)
 [![build](https://github.com/open-telemetry/otel-arrow/actions/workflows/rust-audit.yml/badge.svg)](https://github.com/open-telemetry/otel-arrow/actions/workflows/rust-audit.yml)
@@ -10,7 +10,7 @@
 
 ## Overview
 
-The OTAP Dataflow library is a set of core Rust crates which combine
+The OTAP Dataflow Engine is a set of core Rust crates which combine
 to produce an OpenTelemetry pipeline support, for use as an embedded
 software component, providing a framework for collecting OpenTelemetry
 data.
@@ -337,6 +337,7 @@ docker build --build-context otel-arrow=../../ -f Dockerfile -t df_engine .
 ## Contributing
 
 - [Contribution Guidelines](CONTRIBUTING.md)
+- [Internal Telemetry Guidelines](docs/telemetry/README.md)
 - Code of Conduct (TBD)
 
 Before submitting a PR, please run the following commands:

@@ -299,21 +299,12 @@ impl local::Processor<OtapPdata> for AttributesProcessor {
                 _ => Ok(()),
             },
             Message::PData(pdata) => {
-                if let Some(m) = self.metrics.as_mut() {
-                    m.msgs_consumed.inc();
-                }
-
                 // Fast path: no actions to apply
                 if self.is_noop() {
                     let res = effect_handler
                         .send_message(pdata)
                         .await
                         .map_err(|e| e.into());
-                    if res.is_ok() {
-                        if let Some(m) = self.metrics.as_mut() {
-                            m.msgs_forwarded.inc();
-                        }
-                    }
                     return res;
                 }
 
@@ -354,16 +345,10 @@ impl local::Processor<OtapPdata> for AttributesProcessor {
                     }
                 }
 
-                let res = effect_handler
+                effect_handler
                     .send_message(OtapPdata::new(context, records.into()))
                     .await
-                    .map_err(|e| e.into());
-                if res.is_ok() {
-                    if let Some(m) = self.metrics.as_mut() {
-                        m.msgs_forwarded.inc();
-                    }
-                }
-                res
+                    .map_err(|e| e.into())
             }
         }
     }
@@ -1171,8 +1156,6 @@ mod telemetry_tests {
                 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
 
                 // Inspect current metrics; fields with non-zero values should be present
-                let mut found_consumed = false;
-                let mut found_forwarded = false;
                 let mut found_renamed_entries = false;
                 let mut found_deleted_entries = false;
                 let mut found_domain_signal = false;
@@ -1181,8 +1164,6 @@ mod telemetry_tests {
                     if desc.name == "attributes.processor.metrics" {
                         for (field, v) in iter {
                             match (field.name, v.to_u64_lossy()) {
-                                ("msgs.consumed", x) if x >= 1 => found_consumed = true,
-                                ("msgs.forwarded", x) if x >= 1 => found_forwarded = true,
                                 ("renamed.entries", x) if x >= 1 => found_renamed_entries = true,
                                 ("deleted.entries", x) if x >= 1 => found_deleted_entries = true,
                                 ("domains.signal", x) if x >= 1 => found_domain_signal = true,
@@ -1192,8 +1173,6 @@ mod telemetry_tests {
                     }
                 });
 
-                assert!(found_consumed, "msgs.consumed should be >= 1");
-                assert!(found_forwarded, "msgs.forwarded should be >= 1");
                 assert!(found_renamed_entries, "renamed.entries should be >= 1");
                 assert!(found_deleted_entries, "deleted.entries should be >= 1");
                 assert!(found_domain_signal, "domains.signal should be >= 1");

@@ -10,14 +10,6 @@ use otap_df_telemetry_macros::metric_set;
 #[metric_set(name = "attributes.processor.metrics")]
 #[derive(Debug, Default, Clone)]
 pub struct AttributesProcessorMetrics {
-    /// PData messages consumed by this processor.
-    #[metric(unit = "{msg}")]
-    pub msgs_consumed: Counter<u64>,
-
-    /// PData messages forwarded by this processor.
-    #[metric(unit = "{msg}")]
-    pub msgs_forwarded: Counter<u64>,
-
     /// Number of failed transform attempts.
     #[metric(unit = "{op}")]
     pub transform_failed: Counter<u64>,

@@ -173,7 +173,6 @@ impl Processor<OtapPdata> for TransformProcessor {
                 }
             },
             Message::PData(pdata) => {
-                self.metrics.msgs_consumed.inc();
                 let (context, payload) = pdata.into_parts();
                 let payload = if !self.should_process(&payload) {
                     // skip handling this pdata
@@ -200,8 +199,7 @@ impl Processor<OtapPdata> for TransformProcessor {
 
                 effect_handler
                     .send_message(OtapPdata::new(context, payload))
-                    .await
-                    .inspect(|_| self.metrics.msgs_forwarded.inc())?;
+                    .await?;
             }
         };
 
@@ -347,17 +345,13 @@ mod test {
                 // Allow the collector to pull from the channel
                 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
 
-                let mut msgs_consumed = 0;
-                let mut msgs_forwarded = 0;
                 let mut msgs_transformed = 0;
                 let mut msgs_transform_failed = 0;
                 registry.visit_current_metrics(|desc, _attrs, iter| {
-                    if desc.name == "transform.processor.metrics" {
+                    if desc.name == "transform.processor" {
                         for (field, v) in iter {
                             let val = v.to_u64_lossy();
                             match field.name {
-                                "msgs.consumed" => msgs_consumed += val,
-                                "msgs.forwarded" => msgs_forwarded += val,
                                 "msgs.transformed" => msgs_transformed += val,
                                 "msgs.transform.failed" => msgs_transform_failed += val,
                                 _ => {}
@@ -366,8 +360,6 @@ mod test {
                     }
                 });
 
-                assert_eq!(msgs_consumed, 1);
-                assert_eq!(msgs_forwarded, 1);
                 assert_eq!(msgs_transformed, 1);
                 assert_eq!(msgs_transform_failed, 0)
             });

@@ -7,17 +7,9 @@ use otap_df_telemetry::instrument::Counter;
 use otap_df_telemetry_macros::metric_set;
 
 /// Metrics for the TransformProcessor node.
-#[metric_set(name = "transform.processor.metrics")]
+#[metric_set(name = "transform.processor")]
 #[derive(Debug, Default, Clone)]
 pub struct Metrics {
-    /// PData messages consumed by this processor.
-    #[metric(unit = "{msg}")]
-    pub msgs_consumed: Counter<u64>,
-
-    /// PData messages forwarded by this processor.
-    #[metric(unit = "{msg}")]
-    pub msgs_forwarded: Counter<u64>,
-
     /// Number of messages successfully transformed.
     #[metric(unit = "{msg}")]
     pub msgs_transformed: Counter<u64>,

@@ -1,5 +1,7 @@
 # Telemetry SDK (schema-first, multivariate, NUMA-aware)
 
+Status: draft under active development.
+
 A low-overhead, NUMA-aware telemetry SDK that turns a declarative schema into a
 type-safe Rust API for emitting richly structured, multivariate metrics. It is
 designed for engines that run a thread-per-core and require predictable latency
@@ -62,3 +64,6 @@ See the [telemetry-macros crate](../telemetry-macros) for details.
 - NUMA-aware aggregation.
 
 ![Architecture Phase 2](assets/Metrics%20Phase%202.svg)
+
+Note: The recent telemetry guidelines defined in `/docs/telemetry` are
+still being implemented in this SDK. Expect changes and improvements over time.