OTLP Exporter Optimizations (open-telemetry#1474)

Support for multiple simultaneous client connections to improve
throughput and the ack/nack system.

---------

Co-authored-by: Joshua MacDonald <jmacd@users.noreply.github.com>

Author: lquerel

rust/otap-dataflow/Cargo.toml

@@ -68,16 +68,16 @@ flume = { version = "0.11.1", default-features = false, features = ["async"] }
 futures = "0.3.31"
 futures-channel = "0.3"
 futures-timer = "3.0"
-http = "1.3"
-humantime = "2.2.0"
+http = "1.4"
+humantime = "2.3.0"
 humantime-serde = "1.1.1"
 itertools = "0.14.0"
-linkme = "0.3.33"
+linkme = "0.3.35"
 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"] }
+nix = { version = "0.30.1", features = ["process", "signal"] }
 num_enum = "0.7"
 object_store = "0.12.3"
 once_cell = "1.20.2"
@@ -102,13 +102,13 @@ replace_with = "0.1.8"
 simdutf8 = "0.1.5"
 slotmap = "1.0.7"
 smallvec = { version = "1.15" , features = ["union"] }
-socket2 = { version = "0.6.0", features = ["all"] }
+socket2 = { version = "0.6.1", 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"] }
+thiserror = "2.0.17"
+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",

rust/otap-dataflow/crates/otap/src/metrics.rs

@@ -55,26 +55,47 @@ pub struct ExporterPDataMetrics {
 
 impl ExporterPDataMetrics {
     pub fn inc_consumed(&mut self, st: SignalType) {
+        self.add_consumed(st, 1);
+    }
+
+    pub fn inc_exported(&mut self, st: SignalType) {
+        self.add_exported(st, 1);
+    }
+
+    pub fn inc_failed(&mut self, st: SignalType) {
+        self.add_failed(st, 1);
+    }
+
+    pub fn add_consumed(&mut self, st: SignalType, count: u64) {
+        if count == 0 {
+            return;
+        }
         match st {
-            SignalType::Metrics => self.metrics_consumed.inc(),
-            SignalType::Logs => self.logs_consumed.inc(),
-            SignalType::Traces => self.traces_consumed.inc(),
+            SignalType::Metrics => self.metrics_consumed.add(count),
+            SignalType::Logs => self.logs_consumed.add(count),
+            SignalType::Traces => self.traces_consumed.add(count),
         }
     }
 
-    pub fn inc_exported(&mut self, st: SignalType) {
+    pub fn add_exported(&mut self, st: SignalType, count: u64) {
+        if count == 0 {
+            return;
+        }
         match st {
-            SignalType::Metrics => self.metrics_exported.inc(),
-            SignalType::Logs => self.logs_exported.inc(),
-            SignalType::Traces => self.traces_exported.inc(),
+            SignalType::Metrics => self.metrics_exported.add(count),
+            SignalType::Logs => self.logs_exported.add(count),
+            SignalType::Traces => self.traces_exported.add(count),
         }
     }
 
-    pub fn inc_failed(&mut self, st: SignalType) {
+    pub fn add_failed(&mut self, st: SignalType, count: u64) {
+        if count == 0 {
+            return;
+        }
         match st {
-            SignalType::Metrics => self.metrics_failed.inc(),
-            SignalType::Logs => self.logs_failed.inc(),
-            SignalType::Traces => self.traces_failed.inc(),
+            SignalType::Metrics => self.metrics_failed.add(count),
+            SignalType::Logs => self.logs_failed.add(count),
+            SignalType::Traces => self.traces_failed.add(count),
         }
     }
 }

rust/otap-dataflow/crates/otap/src/otap_grpc.rs

@@ -32,6 +32,7 @@ use crate::{
     pdata::{Context, OtapPdata},
 };
 
+pub mod client_settings;
 pub mod middleware;
 pub mod otlp;
 pub mod server_settings;

rust/otap-dataflow/crates/otap/src/otap_grpc/client_settings.rs

@@ -18,7 +18,7 @@ pub struct GrpcClientSettings {
     pub grpc_endpoint: String,
 
     /// Compression method to use for outbound requests. Defaults to no compression.
-    #[serde(default)]
+    #[serde(default, alias = "compression_method")]
     pub compression: Option<CompressionMethod>,
 
     /// Maximum number of concurrent in-flight requests allowed by the transport stack.
@@ -162,35 +162,35 @@ const fn default_concurrency_limit() -> usize {
     256
 }
 
-fn default_connect_timeout() -> Duration {
+const fn default_connect_timeout() -> Duration {
     Duration::from_secs(3)
 }
 
 const fn default_tcp_nodelay() -> bool {
     true
 }
 
-fn default_tcp_keepalive() -> Option<Duration> {
+const fn default_tcp_keepalive() -> Option<Duration> {
     Some(Duration::from_secs(45))
 }
 
-fn default_initial_stream_window_size() -> Option<u32> {
+const fn default_initial_stream_window_size() -> Option<u32> {
     Some(8 * 1024 * 1024)
 }
 
-fn default_initial_connection_window_size() -> Option<u32> {
+const fn default_initial_connection_window_size() -> Option<u32> {
     Some(32 * 1024 * 1024)
 }
 
 const fn default_http2_adaptive_window() -> bool {
     false
 }
 
-fn default_http2_keepalive_interval() -> Option<Duration> {
+const fn default_http2_keepalive_interval() -> Option<Duration> {
     Some(Duration::from_secs(30))
 }
 
-fn default_http2_keepalive_timeout() -> Option<Duration> {
+const fn default_http2_keepalive_timeout() -> Option<Duration> {
     Some(Duration::from_secs(10))
 }
 

rust/otap-dataflow/crates/otap/src/otap_grpc/server_settings.rs

@@ -253,43 +253,43 @@ const fn default_tcp_nodelay() -> bool {
     true
 }
 
-fn default_tcp_keepalive() -> Option<Duration> {
+const fn default_tcp_keepalive() -> Option<Duration> {
     Some(Duration::from_secs(45))
 }
 
-fn default_tcp_keepalive_interval() -> Option<Duration> {
+const fn default_tcp_keepalive_interval() -> Option<Duration> {
     Some(Duration::from_secs(15))
 }
 
-fn default_tcp_keepalive_retries() -> Option<u32> {
+const fn default_tcp_keepalive_retries() -> Option<u32> {
     Some(5)
 }
 
 const fn default_load_shed() -> bool {
     true
 }
 
-fn default_initial_stream_window_size() -> Option<u32> {
+const fn default_initial_stream_window_size() -> Option<u32> {
     Some(8 * 1024 * 1024)
 }
 
-fn default_initial_connection_window_size() -> Option<u32> {
+const fn default_initial_connection_window_size() -> Option<u32> {
     Some(24 * 1024 * 1024)
 }
 
-fn default_max_frame_size() -> Option<u32> {
+const fn default_max_frame_size() -> Option<u32> {
     Some(16 * 1024)
 }
 
-fn default_max_decoding_message_size() -> Option<u32> {
+const fn default_max_decoding_message_size() -> Option<u32> {
     Some(4 * 1024 * 1024)
 }
 
-fn default_http2_keepalive_interval() -> Option<Duration> {
+const fn default_http2_keepalive_interval() -> Option<Duration> {
     Some(Duration::from_secs(30))
 }
 
-fn default_http2_keepalive_timeout() -> Option<Duration> {
+const fn default_http2_keepalive_timeout() -> Option<Duration> {
     Some(Duration::from_secs(10))
 }