feat(otlp): add gRPC protocol support for trace ingestion (#1105)

## Summary

- Add gRPC support for OTLP trace ingestion on port 4317, complementing
the existing HTTP endpoint on port 4318
- Implement `TraceService` trait using tonic for gRPC server
- Refactor `OtlpProcessor` to accept `ExportTraceServiceRequest`
directly (avoids re-encoding for gRPC)
- Wire up existing config:
`DD_OTLP_CONFIG_RECEIVER_PROTOCOLS_GRPC_ENDPOINT` and
`DD_OTLP_CONFIG_RECEIVER_PROTOCOLS_GRPC_MAX_RECV_MSG_SIZE_MIB`

## Motivation

Users were configuring gRPC OTLP endpoints
(`DD_OTLP_CONFIG_RECEIVER_PROTOCOLS_GRPC_ENDPOINT`) but the feature was
not implemented, resulting in silent failures. This PR enables gRPC
support. This was for an old ticket, but we should go ahead and add
this, especially since OTLP is getting more popular.

## Changes

| File | Change |
|------|--------|
| `bottlecap/Cargo.toml` | Add tonic with transport/server features |
| `bottlecap/src/otlp/grpc_agent.rs` | NEW - gRPC server implementing
TraceService |
| `bottlecap/src/otlp/processor.rs` | Refactor to accept
ExportTraceServiceRequest directly |
| `bottlecap/src/otlp/mod.rs` | Update enablement logic for HTTP/gRPC |
| `bottlecap/src/bin/bottlecap/main.rs` | Start both HTTP and gRPC
servers |

## Binary Size Impact

Measured from CI layer size checks (arm64):

| Metric | Main | PR | Difference |
|--------|------|-----|------------|
| Zipped | 5,074 KB | 5,164 KB | **+90 KB** |
| Unzipped | 11,183 KB | 11,376 KB | **+193 KB** |

This is a ~1.7% increase, well under historical concerns (OTLP was
removed from Go agent for adding ~1.15MB).

## Test plan

- [x] New integration tests for gRPC protocol (4 tests)
- [x] All existing OTLP tests continue to pass
- [x] `cargo fmt`, `cargo clippy`, `cargo test` pass
- [x] All CI checks passed (47 checks including integration suites)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: jchrostek-dd

bottlecap/Cargo.lock

@@ -45,6 +45,7 @@ rustls-pki-types = { version = "1.0", default-features = false }
 hyper-rustls = { version = "0.27.7", default-features = false }
 rand = { version = "0.8", default-features = false }
 prost = { version = "0.14", default-features = false }
+tonic = { version = "0.14", features = ["transport", "codegen", "server", "channel", "router"], default-features = false }
 tonic-types = { version = "0.14", default-features = false }
 zstd = { version = "0.13.3", default-features = false }
 futures = { version = "0.3.31", default-features = false }

bottlecap/Cargo.toml

@@ -52,7 +52,10 @@ use bottlecap::{
         },
         flusher::LogsFlusher,
     },
-    otlp::{agent::Agent as OtlpAgent, should_enable_otlp_agent},
+    otlp::{
+        agent::Agent as OtlpAgent, grpc_agent::GrpcAgent as OtlpGrpcAgent, should_enable_otlp_grpc,
+        should_enable_otlp_http,
+    },
     proxy::{interceptor, should_start_proxy},
     secrets::decrypt,
     tags::{
@@ -1354,24 +1357,49 @@ fn start_otlp_agent(
     trace_tx: Sender<SendDataBuilderInfo>,
     stats_concentrator: StatsConcentratorHandle,
 ) -> Option<CancellationToken> {
-    if !should_enable_otlp_agent(config) {
+    let http_enabled = should_enable_otlp_http(config);
+    let grpc_enabled = should_enable_otlp_grpc(config);
+
+    if !http_enabled && !grpc_enabled {
         return None;
     }
+
     let stats_generator = Arc::new(StatsGenerator::new(stats_concentrator));
-    let agent = OtlpAgent::new(
-        config.clone(),
-        tags_provider,
-        trace_processor,
-        trace_tx,
-        stats_generator,
-    );
-    let cancel_token = agent.cancel_token();
+    let cancel_token = CancellationToken::new();
 
-    tokio::spawn(async move {
-        if let Err(e) = agent.start().await {
-            error!("Error starting OTLP agent: {e:?}");
-        }
-    });
+    if http_enabled {
+        let agent = OtlpAgent::new(
+            config.clone(),
+            tags_provider.clone(),
+            trace_processor.clone(),
+            trace_tx.clone(),
+            stats_generator.clone(),
+            cancel_token.clone(),
+        );
+
+        tokio::spawn(async move {
+            if let Err(e) = agent.start().await {
+                error!("Error starting OTLP HTTP agent: {e:?}");
+            }
+        });
+    }
+
+    if grpc_enabled {
+        let grpc_agent = OtlpGrpcAgent::new(
+            config.clone(),
+            tags_provider,
+            trace_processor,
+            trace_tx,
+            stats_generator,
+            cancel_token.clone(),
+        );
+
+        tokio::spawn(async move {
+            if let Err(e) = grpc_agent.start().await {
+                error!("Error starting OTLP gRPC agent: {e:?}");
+            }
+        });
+    }
 
     Some(cancel_token)
 }

bottlecap/src/bin/bottlecap/main.rs

@@ -56,12 +56,12 @@ impl Agent {
         trace_processor: Arc<dyn TraceProcessor + Send + Sync>,
         trace_tx: Sender<SendDataBuilderInfo>,
         stats_generator: Arc<StatsGenerator>,
+        cancel_token: CancellationToken,
     ) -> Self {
         let port = Self::parse_port(
             config.otlp_config_receiver_protocols_http_endpoint.as_ref(),
             OTLP_AGENT_HTTP_PORT,
         );
-        let cancel_token = CancellationToken::new();
 
         Self {
             config: Arc::clone(&config),
@@ -75,11 +75,6 @@ impl Agent {
         }
     }
 
-    #[must_use]
-    pub fn cancel_token(&self) -> CancellationToken {
-        self.cancel_token.clone()
-    }
-
     fn parse_port(endpoint: Option<&String>, default_port: u16) -> u16 {
         if let Some(endpoint) = endpoint {
             let port = endpoint.split(':').nth(1);
@@ -312,7 +307,6 @@ mod tests {
 
     #[test]
     fn test_parse_port_with_valid_endpoint() {
-        // Test with a valid endpoint containing a port
         let endpoint = Some("localhost:8080".to_string());
         assert_eq!(
             Agent::parse_port(endpoint.as_ref(), OTLP_AGENT_HTTP_PORT),
@@ -322,7 +316,6 @@ mod tests {
 
     #[test]
     fn test_parse_port_with_invalid_port_format() {
-        // Test with an endpoint containing an invalid port format
         let endpoint = Some("localhost:invalid".to_string());
         assert_eq!(
             Agent::parse_port(endpoint.as_ref(), OTLP_AGENT_HTTP_PORT),
@@ -332,7 +325,6 @@ mod tests {
 
     #[test]
     fn test_parse_port_with_missing_port() {
-        // Test with an endpoint missing a port
         let endpoint = Some("localhost".to_string());
         assert_eq!(
             Agent::parse_port(endpoint.as_ref(), OTLP_AGENT_HTTP_PORT),
@@ -342,7 +334,6 @@ mod tests {
 
     #[test]
     fn test_parse_port_with_none_endpoint() {
-        // Test with None endpoint
         let endpoint: Option<String> = None;
         assert_eq!(
             Agent::parse_port(endpoint.as_ref(), OTLP_AGENT_HTTP_PORT),
@@ -352,7 +343,6 @@ mod tests {
 
     #[test]
     fn test_parse_port_with_empty_endpoint() {
-        // Test with an empty endpoint
         let endpoint = Some(String::new());
         assert_eq!(
             Agent::parse_port(endpoint.as_ref(), OTLP_AGENT_HTTP_PORT),