Second round setup (open-telemetry#343)

Key Changes:

- Definition of the `Exporter` trait
- Alignment of the `Processor` trait signature with other traits
- Creation of a testing infrastructure for the three node types and
adaptation of existing tests to this new framework
- Added documentation on pipeline definition

---------

Co-authored-by: Utkarsh Umesan Pillai <66651184+utpilla@users.noreply.github.com>

Author: lquerel

rust/otap-dataflow/.clippy.toml

@@ -4,4 +4,21 @@ allow-expect-in-tests = true
 allow-unwrap-in-tests = true
 allow-dbg-in-tests = true
 allow-print-in-tests = true
-disallowed-methods = []
+
+# Disallow specific methods
+disallowed-methods = []
+
+# Disallow specific types
+disallowed-types = [
+    { path = "once_cell::sync::Lazy", reason = "Please use `std::sync::LazyLock` instead." },
+]
+
+# Disallow specific macros
+disallowed-macros = [
+    { path = "lazy_static::lazy_static", reason = "Please use `std::sync::LazyLock` instead." },
+]
+
+doc-valid-idents = [
+    "OTEL",
+    "..",   # add the default list
+]

rust/otap-dataflow/.cursor/rules/single-threaded-async-runtime.mdc

@@ -6,7 +6,7 @@ alwaysApply: true
 # Design Principles
 
 - Target a single-threaded async runtime
-- Favor ?Send futures whenever possible
+- Declare async traits as `?Send`, providing `!Send` implementations and futures whenever practical
 - Avoid synchronization primitives as much as possible
 - Optimize for performance
 - Avoid unbounded channels and data structures

rust/otap-dataflow/ROADMAP.md

@@ -27,16 +27,31 @@ significant changes.
 - [ ] Continuous benchmarking and performance tracking
   - [ ] Comparison with Go Collector
 
-### Pipeline Engine Foundations
+### Pipeline Engine Foundations (phase 1)
 
 - Channels
   - [x] MPMC Channel
   - [x] MPSC Channel
 - [x] Receiver trait
-- [WIP] Processor trait
-- [ ] Exporter trait
+- [x] Processor trait
+- [x] Exporter trait
 - [ ] Pipeline Engine
-- [ ] Initial benchmarks and documentation
+- [ ] CPU & Memory Benchmarks
+- [ ] Documentation
+
+### Pipeline Engine Foundations (phase 2)
+
+- Channels
+  - [ ] SPSC Channel
+  - [ ] Broadcast Channel
+- [ ] Connector trait
+- [ ] Instrumentation
+- [ ] Thread pinning
+- Benchmarks
+  - [ ] CPU benchmarks
+  - [ ] Memory benchmarks
+- [ ] CPU & Memory Benchmarks
+- [ ] Documentation
 
 ### OTLP Pipeline
 

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

@@ -13,5 +13,5 @@ rust-version.workspace = true
 workspace = true
 
 [dependencies]
-thiserror.workspace = true
-tokio.workspace = true
+thiserror = { workspace = true }
+tokio = { workspace = true }

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

@@ -13,6 +13,6 @@ rust-version.workspace = true
 workspace = true
 
 [dependencies]
-thiserror.workspace = true
-serde.workspace = true
-serde_json.workspace = true
+thiserror = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }

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

@@ -16,9 +16,9 @@ workspace = true
 otap-df-channel = { path = "../channel" }
 otap-df-config = { path = "../config" }
 
-thiserror.workspace = true
-serde_json.workspace = true
-tokio.workspace = true
-async-trait.workspace = true
+thiserror = { workspace = true }
+serde_json = { workspace = true }
+tokio = { workspace = true }
+async-trait = { workspace = true }
 
 socket2 = "0.5.8"

rust/otap-dataflow/crates/engine/README.md

@@ -2,46 +2,83 @@
 
 Status: **WIP**
 
-## Nodes
-
-The Pipeline Engine consists of three main node types:
-
-- Receiver: A node that receives or scrapes telemetry data from a telemetry
-  source.
-- Processor: A node that processes telemetry data.
-- Exporter: A node that exports telemetry data to a telemetry sink.
-
-Each node can receive control messages from the Pipeline Engine to manage its
-operation.
-
-```mermaid
-graph LR
-    RecvControlIn[Control Channel] --> Receiver
-    Receiver((Receiver))
-    Receiver --> RecvControlOut[Control Channel]
-    Receiver --> RecvDataOut[Data Channel]
-    class Receiver node
-    class RecvControlIn,RecvControlOut controlQueue
-    class RecvDataOut dataQueue
-    
-    ProcControlIn[Control Channel] --> Processor
-    ProcDataIn[Data Channel] --> Processor
-    Processor((Processor))
-    Processor --> ProcControlOut[Control Channel]
-    Processor --> ProcDataOut[Data Channel]
-    class Processor node
-    class ProcControlIn,ProcControlOut controlQueue
-    class ProcDataIn,ProcDataOut dataQueue
-
-    ExControlIn[Control Channel] --> Exporter
-    ExDataIn[Data Channel] --> Exporter
-    Exporter((Exporter))
-    Exporter --> ExControlOut[Control Channel]
-    class Exporter node
-    class ExControlIn,ExControlOut controlQueue
-    class ExDataIn,ExDataOut dataQueue
-
-    classDef controlQueue fill:#ffdddd,stroke:#cc0000,color:#000,stroke-width:0px,font-size:10px,padding:0px
-    classDef dataQueue fill:#ddffdd,stroke:#009900,color:#000,stroke-width:0px,font-size:10px,padding:0px
-    classDef node fill:#4a90e2,color:#ffffff,stroke-width:0px,font-size:12px
-```
+## Introduction
+
+The term pipeline is used here to represent an interconnection of nodes forming
+a Directed Acyclic Graph (DAG). The inputs of a pipeline are called receivers,
+the intermediate processing nodes are called processors, and the output nodes
+are referred to as exporters.
+
+Messages flowing through a pipeline are generically referred to as pdata (i.e.
+Pipeline Data). An OTLP message or an OTAP message are examples of pdata types.
+This pipeline framework is generic over pdata, which means:
+
+- It is possible to instantiate an OTLP pipeline, an OTAP pipeline, or even a
+  pipeline designed to support another type of pdata.
+- It is not possible to support multiple pdata types within a single pipeline.
+  However, a fourth type of component, called a connector, can be used to bridge
+  two pipelines and enable interoperability between different pdata types.
+
+This terminology aligns well with the concepts introduced in the OTEL Go
+Collector.
+
+## Architecture
+
+A list of the design principles followed by this project can be found
+[here](../../docs/design-principles.md). More specifically, the pipeline engine
+implemented in this crate follows a share-nothing, thread-per-core approach.
+In particular, one instance of the pipeline engine is created per core. This
+engine:
+
+- Is based on a single-threaded async runtime
+- Avoids synchronization mechanisms whenever possible
+- Declares async traits as `?Send`, providing `!Send` implementations and
+  futures whenever practical
+- Relies on listening sockets configured with the `SO_REUSEPORT` option,
+  allowing the OS to handle connection load balancing
+- May share immutable data between cores, but ideally only within a single NUMA
+  node
+
+These design principles focus on achieving high performance, predictability, and
+maintainability in an observability gateway implemented in Rust. Targeting a
+single-threaded async runtime reduces complexity, enhances cache locality, and
+lowers overhead. Favoring `!Send` futures and declaring async traits as `?Send`
+maximizes flexibility and allows performance gains by avoiding unnecessary
+synchronization. Minimizing synchronization primitives prevents contention and
+overhead, thus ensuring consistently low latency. Avoiding unbounded channels
+and data structures protects against unpredictable resource consumption,
+maintaining stable performance. Finally, limiting external dependencies reduces
+complexity, security risks, and maintenance effort, further streamlining the
+gateway’s operation and reliability.
+
+## Control Messages
+
+Each node in a pipeline can receive control messages, which must be handled with
+priority. These control messages are issued by a control entity (e.g. a pipeline
+engine) and are used to orchestrate the behavior of pipeline nodes. For example,
+configuring or reconfiguring nodes, coordinating acknowledgment mechanisms,
+stopping a pipeline, and more.
+
+## Testability
+
+All node types, as well as the pipeline engine itself, are designed for isolated
+testing. Practically, this means it's possible to test components like an OTLP
+Receiver independently, without needing to construct an entire pipeline. This
+approach facilitates rapid and precise identification of issues such as memory
+overconsumption, bottlenecks, or logical errors within individual nodes.
+
+The engine provides an extensive `testing` module containing utilities tailored
+to each pipeline component:
+
+- Defined test message types and control message counters for monitoring
+  component behavior.
+- Dedicated test contexts and runtimes specifically built for receivers,
+  processors, and exporters.
+- Single-threaded asynchronous runtime utilities aligned with the engine's
+  non-Send design philosophy.
+- Convenient helper functions for establishing and managing communication
+  channels between components.
+
+These utilities streamline the process of validating individual component
+behaviors, significantly reducing setup effort while enabling comprehensive
+testing.

rust/otap-dataflow/crates/engine/src/error.rs

@@ -24,4 +24,37 @@ pub enum Error<T> {
         /// The error that occurred.
         error: std::io::Error,
     },
+
+    /// A wrapper for the receiver errors.
+    #[error("A receiver error occurred in node {receiver}: {error}")]
+    ReceiverError {
+        /// The name of the receiver that encountered the error.
+        receiver: NodeName,
+
+        /// The error that occurred.
+        /// ToDo We probably need to use a more specific error type here (JSON Node?).
+        error: String,
+    },
+
+    /// A wrapper for the processor errors.
+    #[error("A processor error occurred in node {processor}: {error}")]
+    ProcessorError {
+        /// The name of the processor that encountered the error.
+        processor: NodeName,
+
+        /// The error that occurred.
+        /// ToDo We probably need to use a more specific error type here (JSON Node?).
+        error: String,
+    },
+
+    /// A wrapper for the exporter errors.
+    #[error("An exporter error occurred in node {exporter}: {error}")]
+    ExporterError {
+        /// The name of the exporter that encountered the error.
+        exporter: NodeName,
+
+        /// The error that occurred.
+        /// ToDo We probably need to use a more specific error type here (JSON Node?).
+        error: String,
+    },
 }