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

bottlecap/Cargo.toml

@@ -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/src/bin/bottlecap/main.rs

@@ -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/otlp/agent.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);
@@ -209,13 +204,9 @@ impl Agent {
             }
         }
 
-        // This needs to be after process_traces() because process_traces()
-        // performs obfuscation, and we need to compute stats on the obfuscated traces.
         if compute_trace_stats_on_extension
             && let Err(err) = stats_generator.send(&processed_traces)
         {
-            // Just log the error. We don't think trace stats are critical, so we don't want to
-            // return an error if only stats fail to send.
             error!("OTLP | Error sending traces to the stats concentrator: {err}");
         }
 
@@ -312,7 +303,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 +312,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 +321,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 +330,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 +339,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),