Unify OTLP/OTAP gRPC Configuration and Introduce Experimental Non-Tonic Receiver

rust/otap-dataflow/.gitignore

@@ -40,4 +40,9 @@ output.json
 resolved-schema.yaml
 
 # Ignore the temporary repo dir created by `cargo xtask history` command
-history-temp-repo/
+history-temp-repo/
+
+*.patch
+perf.data
+perf.data.*
+*.strace

rust/otap-dataflow/Cargo.toml

@@ -37,6 +37,8 @@ thiserror.workspace = true
 serde_json.workspace = true
 clap.workspace = true
 mimalloc-rust.workspace = true
+env_logger.workspace = true
+jemallocator = { workspace = true, optional = true }
 
 [workspace.dependencies]
 otap-df-pdata-otlp-macros = { path = "./crates/pdata/src/otlp/macros"}
@@ -64,6 +66,7 @@ criterion = "0.7.0"
 data-encoding = "2.9.0"
 fluke-hpack = "0.3.1"
 flume = { version = "0.11.1", default-features = false, features = ["async"] }
+flate2 = "1.1.5"
 futures = "0.3.31"
 futures-channel = "0.3"
 futures-timer = "3.0"
@@ -76,7 +79,9 @@ local-sync = "0.1.1"
 log = "0.4"
 miette = { version="7.6.0", features = ["fancy"] }
 mimalloc-rust = "0.2.1"
-nix = { version = "0.30.0", features = ["process", "signal"] }
+jemalloc-ctl = "0.5.4"
+jemallocator = "0.5.4"
+nix = { version = "0.30.0", features = ["process", "signal", "resource", "feature"] }
 num_enum = "0.7"
 object_store = "0.12.3"
 once_cell = "1.20.2"
@@ -105,9 +110,9 @@ socket2 = { version = "0.6.0", features = ["all"] }
 syn = { version = "2.0", features = ["full", "extra-traits"] }
 tempfile = "3"
 thiserror = "2.0.12"
-tokio = { version = "1.46.1", features = ["rt", "time", "net", "io-util", "sync", "macros", "rt-multi-thread", "fs", "io-std", "process"] }
+tokio = { version = "1.48.0", features = ["rt", "time", "net", "io-util", "sync", "macros", "rt-multi-thread", "fs", "io-std", "process"] }
 tokio-stream = "0.1.17"
-tokio-util = { version = "0.7.16" }
+tokio-util = { version = "0.7.17" }
 tonic = { version = "0.14", default-features = false, features = [
     "channel",
     "codegen",
@@ -133,15 +138,22 @@ weaver_resolved_schema = { git = "https://github.com/open-telemetry/weaver.git",
 weaver_resolver = { git = "https://github.com/open-telemetry/weaver.git", tag = "v0.17.0"}
 weaver_semconv = { git = "https://github.com/open-telemetry/weaver.git", tag = "v0.17.0"}
 zip = "=4.2.0"
+byte-unit = "5.2.0"
+bytes = "1.10.1"
+env_logger = "0.11"
+h2 = "0.4.7"
+zstd = "0.13"
 
 [features]
 default = []
-unsafe-optimizations = ["unchecked-index", "unchecked-arithmetic"]
+unsafe-optimizations = ["unchecked-index", "unchecked-arithmetic", "otap-df-otap/unsafe-optimizations"]
 unchecked-index = []
 unchecked-arithmetic = []
 # Experimental exporters (opt-in)
 experimental-exporters = ["otap-df-otap/experimental-exporters"]
 geneva-exporter = ["otap-df-otap/geneva-exporter"]
+jemalloc-metrics = ["otap-df-otap/jemalloc-metrics"]
+jemalloc-allocator = ["dep:jemallocator"]
 
 [workspace.lints.rust]
 # General compatibility lints
@@ -201,6 +213,12 @@ missing_crate_level_docs = "deny"
 
 [profile.release]
 debug = "line-tables-only"  # minimum required for profiling
+#opt-level = 3
+#lto = "thin"
+#codegen-units = 1
+#panic = "abort"
+#incremental = false
+#strip = "debuginfo"
 
 # A more in-depth analysis is necessary to determine the optimal parameters for the release profile.
 #[profile.release]

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

@@ -413,7 +413,7 @@ fn bench_exporter(c: &mut Criterion) {
             |b, otap_signals| {
                 b.to_async(&rt).iter(|| async {
                     // start perf exporter
-                    let config = Config::new(1000, 0.3, true, true, true, true, true);
+                    let config = Config::new(1000, 0.3, true, true, true, true, true, false);
                     let exporter_config = ExporterConfig::new("perf_exporter");
                     let node_config =
                         Arc::new(NodeUserConfig::new_exporter_config(OTAP_PERF_EXPORTER_URN));
@@ -473,7 +473,7 @@ fn bench_exporter(c: &mut Criterion) {
             |b, otap_signals| {
                 b.to_async(&rt).iter(|| async {
                     // start perf exporter
-                    let config = Config::new(1000, 0.3, false, false, false, false, false);
+                    let config = Config::new(1000, 0.3, false, false, false, false, false, false);
                     let exporter_config = ExporterConfig::new("perf_exporter");
                     let node_config =
                         Arc::new(NodeUserConfig::new_exporter_config(OTAP_PERF_EXPORTER_URN));

rust/otap-dataflow/configs/fake-otlp.yaml

@@ -14,14 +14,13 @@ nodes:
         dispatch_strategy: round_robin
     config:
       traffic_config:
-        signals_per_second: 100000
-        max_signal_count: null
-        metric_weight: 0
-        trace_weight: 0
-        log_weight: 30
+        max_batch_size: 1000
+        signals_per_second: 75000
+        log_weight: 100
       registry_path: https://github.com/open-telemetry/semantic-conventions.git[model]
   exporter:
     kind: exporter
     plugin_urn: "urn:otel:otlp:exporter"
     config:
-      grpc_endpoint: "http://127.0.0.1:4317"
+      grpc_endpoint: "http://127.0.0.1:4327"
+      #compression_method: zstd

rust/otap-dataflow/configs/otlp-otlp.yaml

@@ -6,18 +6,20 @@ settings:
 nodes:
   receiver:
     kind: receiver
-    plugin_urn: "urn:otel:otlp:receiver"
+    plugin_urn: "urn:otel:otel:receiver"
     out_ports:
       out_port:
         destinations:
           - exporter
         dispatch_strategy: round_robin
     config:
-      listening_addr: "127.0.0.1:4317"
+      listening_addr: "127.0.0.1:4327"
+      request_compression: none
       # timeout: "30s"  # Optional: timeout for RPC requests
   exporter:
     kind: exporter
     plugin_urn: "urn:otel:otlp:exporter"
     config:
-      grpc_endpoint: "http://127.0.0.1:4318"
+      grpc_endpoint: "http://127.0.0.1:4328"
+      #compression_method: zstd
       # timeout: "15s"  # Optional: timeout for RPC requests

rust/otap-dataflow/configs/otlp-perf.yaml

@@ -6,20 +6,23 @@ settings:
 nodes:
   receiver:
     kind: receiver
-    plugin_urn: "urn:otel:otlp:receiver"
+    plugin_urn: "urn:otel:otel:receiver"
     out_ports:
       out_port:
         destinations:
           - perf
         dispatch_strategy: round_robin
     config:
-      listening_addr: "127.0.0.1:4318"
+      listening_addr: "127.0.0.1:4328"
+      request_compression: none
+      timeout: "10s"
   perf:
     kind: exporter
     plugin_urn: "urn:otel:otap:perf:exporter"
     config:
       frequency: 1000
-      cpu_usage: false
-      mem_usage: false
+      display_throughput: true
+      cpu_usage: true
+      mem_usage: true
       disk_usage: false
       io_usage: false

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

@@ -20,3 +20,4 @@ serde_yaml = { workspace = true }
 miette = { workspace = true }
 urn = { workspace = true }
 schemars = { workspace = true }
+byte-unit = { workspace = true }

rust/otap-dataflow/crates/config/src/byte_units.rs

@@ -0,0 +1,127 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Support for byte units like "KB / KiB", "MB / MiB", "GB / GiB" in configuration files.
+
+use byte_unit::Byte;
+use serde::de::Error as DeError;
+use serde::{Deserialize, Deserializer};
+
+#[derive(Deserialize)]
+#[serde(untagged)]
+enum Value {
+    Number(u64),
+    String(String),
+}
+
+/// Deserialize an optional byte size that can be specified either as a number (in bytes)
+/// or as a string with units (e.g. "1 KB", "2 MiB").
+pub fn deserialize<'de, D>(deserializer: D) -> Result<Option<u32>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value = Option::<Value>::deserialize(deserializer)?;
+    let Some(value) = value else {
+        return Ok(None);
+    };
+
+    let (bytes, repr) = match value {
+        Value::Number(value) => (value as u128, value.to_string()),
+        Value::String(text) => {
+            let parsed: Byte = text.parse().map_err(DeError::custom)?;
+            (parsed.as_u64() as u128, text)
+        }
+    };
+
+    if bytes > u32::MAX as u128 {
+        return Err(DeError::custom(format!(
+            "byte size '{}' ({} bytes) exceeds u32::MAX ({} bytes)",
+            repr,
+            bytes,
+            u32::MAX
+        )));
+    }
+
+    Ok(Some(bytes as u32))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::deserialize;
+    use serde::Deserialize;
+
+    #[derive(Debug, Deserialize)]
+    struct Holder {
+        #[serde(default, deserialize_with = "deserialize")]
+        value: Option<u32>,
+    }
+
+    fn de_yaml(input: &str) -> Result<Holder, serde_yaml::Error> {
+        serde_yaml::from_str::<Holder>(input)
+    }
+
+    #[test]
+    fn parses_number_as_bytes() {
+        let cfg = de_yaml("value: 1024").expect("should parse numeric bytes");
+        assert_eq!(cfg.value, Some(1024));
+    }
+
+    #[test]
+    fn parses_string_with_iec_units() {
+        // 1 KiB == 1024 bytes
+        let cfg = de_yaml("value: 1 KiB").expect("should parse 1 KiB");
+        assert_eq!(cfg.value, Some(1024));
+
+        // 2 MiB == 2 * 1024 * 1024 bytes
+        let cfg = de_yaml("value: '2 MiB'").expect("should parse 2 MiB");
+        assert_eq!(cfg.value, Some(2 * 1024 * 1024));
+    }
+
+    #[test]
+    fn parses_plain_string_number() {
+        let cfg = de_yaml("value: '2048'").expect("should parse plain numeric string");
+        assert_eq!(cfg.value, Some(2048));
+    }
+
+    #[test]
+    fn missing_value_is_none() {
+        let cfg = de_yaml("{}").expect("should parse with missing field as None");
+        assert_eq!(cfg.value, None);
+    }
+
+    #[test]
+    fn overflow_is_rejected() {
+        // 4 GiB == 4 * 1024^3 bytes = 4_294_967_296 > u32::MAX (4_294_967_295)
+        let err = de_yaml("value: 4 GiB").expect_err("should error for overflow");
+        let msg = err.to_string();
+        assert!(
+            msg.contains("exceeds u32::MAX"),
+            "unexpected error: {}",
+            msg
+        );
+    }
+
+    #[test]
+    fn parses_no_space_decimal_units() {
+        let cfg = de_yaml("value: 1KB").expect("should parse 1KB without space");
+        assert_eq!(cfg.value, Some(1000));
+
+        let cfg = de_yaml("value: 10MB").expect("should parse 10MB without space");
+        assert_eq!(cfg.value, Some(10_000_000));
+
+        // Lowercase 'b' should still be treated as bytes per crate behavior
+        let cfg = de_yaml("value: 1kb").expect("should parse 1kb as 1000 bits => 125 bytes");
+        assert_eq!(cfg.value, Some(125));
+    }
+
+    #[test]
+    fn parses_fractional_values_and_rounding() {
+        // Decimal unit with fraction
+        let cfg = de_yaml("value: '1.5 MB'").expect("should parse 1.5 MB");
+        assert_eq!(cfg.value, Some(1_500_000));
+
+        // Binary unit with fraction (exact)
+        let cfg = de_yaml("value: '0.5 KiB'").expect("should parse 0.5 KiB to 512 bytes");
+        assert_eq!(cfg.value, Some(512));
+    }
+}

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

@@ -15,6 +15,7 @@
 use serde::{Deserialize, Serialize};
 use std::borrow::Cow;
 
+pub mod byte_units;
 pub mod engine;
 pub mod error;
 pub mod health;