fix(opentelemetry source): bound in-flight request processing

changelog.d/otlp_request_concurrency.security.md

@@ -0,0 +1,3 @@
+The `opentelemetry` source now applies in-flight request controls to prevent unbounded resource consumption.
+
+authors: Jansen-w

src/sources/opentelemetry/config.rs

@@ -1,4 +1,8 @@
-use std::net::SocketAddr;
+use std::{
+    net::SocketAddr,
+    num::{NonZeroU64, NonZeroUsize},
+    time::Duration,
+};
 
 use crate::{
     config::{
@@ -16,7 +20,8 @@ use crate::{
         },
         util::{
             decompression::max_decompressed_size_bytes,
-            grpc::{GrpcKeepaliveConfig, run_grpc_server_with_routes},
+            grpc::{GrpcKeepaliveConfig, run_grpc_server_with_routes_and_request_limiter},
+            request_limiter::{MAX_IN_FLIGHT_EVENTS_TARGET, RequestLimiter},
         },
     },
 };
@@ -118,6 +123,29 @@ pub struct OpentelemetryConfig {
 
     pub http: HttpConfig,
 
+    /// The maximum number of HTTP and gRPC requests that can be processed concurrently.
+    ///
+    /// The adaptive limit is shared across both protocols. It starts with up to two concurrent
+    /// requests, adapts based on observed request sizes, and never exceeds this maximum.
+    /// Requests above the current limit are rejected
+    /// immediately with an overload response.
+    ///
+    /// When unset, the maximum defaults to Vector's configured runtime worker count, falling back
+    /// to the detected available parallelism.
+    #[serde(default)]
+    #[configurable(metadata(docs::examples = 8))]
+    pub max_concurrent_requests: Option<NonZeroUsize>,
+
+    /// The maximum time to cooperatively wait for an admitted HTTP or gRPC request.
+    ///
+    /// The timeout covers receiving and decoding the request body, sending events to the pipeline,
+    /// and waiting for end-to-end acknowledgements. Timed-out requests receive a retryable response
+    /// and release their concurrency permit. Synchronous decompression and decoding cannot be
+    /// preempted, so a request can exceed this duration.
+    #[serde(default = "default_request_timeout_secs")]
+    #[configurable(metadata(docs::examples = 30, docs::type_unit = "seconds"))]
+    pub request_timeout_secs: NonZeroU64,
+
     #[serde(default, deserialize_with = "bool_or_struct")]
     pub acknowledgements: SourceAcknowledgementsConfig,
 
@@ -228,11 +256,23 @@ fn example_http_config() -> HttpConfig {
     }
 }
 
+pub(super) const fn default_request_timeout_secs() -> NonZeroU64 {
+    NonZeroU64::new(30).expect("30 is nonzero")
+}
+
+pub(super) fn default_max_concurrent_requests() -> usize {
+    crate::app::worker_threads()
+        .map(NonZeroUsize::get)
+        .unwrap_or_else(crate::num_threads)
+}
+
 impl GenerateConfig for OpentelemetryConfig {
     fn generate_config() -> serde_json::Value {
         serde_json::to_value(Self {
             grpc: example_grpc_config(),
             http: example_http_config(),
+            max_concurrent_requests: None,
+            request_timeout_secs: default_request_timeout_secs(),
             acknowledgements: Default::default(),
             log_namespace: None,
             use_otlp_decoding: OtlpDecodingConfig::default(),
@@ -273,6 +313,13 @@ impl OpentelemetryConfig {
         let acknowledgements = cx.do_acknowledgements(self.acknowledgements);
         let events_received = register!(EventsReceived);
         let log_namespace = cx.log_namespace(self.log_namespace);
+        let max_concurrent_requests = self
+            .max_concurrent_requests
+            .map(NonZeroUsize::get)
+            .unwrap_or_else(default_max_concurrent_requests);
+        let request_timeout = Duration::from_secs(self.request_timeout_secs.get());
+        let request_limiter =
+            RequestLimiter::new(MAX_IN_FLIGHT_EVENTS_TARGET, max_concurrent_requests);
 
         let grpc_tls_settings = MaybeTlsSettings::from_config(self.grpc.tls.as_ref(), true)?;
 
@@ -326,13 +373,15 @@ impl OpentelemetryConfig {
             .add_service(metrics_service)
             .add_service(trace_service);
 
-        let grpc_source = run_grpc_server_with_routes(
+        let grpc_source = run_grpc_server_with_routes_and_request_limiter(
             self.grpc.address,
             grpc_tls_settings,
             grpc_tls_reloader,
             builder.routes(),
             self.grpc.keepalive.clone(),
             cx.shutdown.clone(),
+            request_limiter.clone(),
+            request_timeout,
         )
         .map_err(|error| {
             error!(message = "OpenTelemetry source gRPC server failed.", %error);
@@ -363,6 +412,8 @@ impl OpentelemetryConfig {
             filters,
             cx.shutdown,
             self.http.keepalive.clone(),
+            request_limiter,
+            request_timeout,
         )
         .map_err(|error| {
             error!(message = "OpenTelemetry source HTTP server failed.", %error);

src/sources/opentelemetry/grpc.rs

@@ -25,7 +25,10 @@ use vector_lib::{
 use crate::{
     SourceSender,
     internal_events::{EventsReceived, StreamClosedError},
-    sources::opentelemetry::config::{LOGS, METRICS, TRACES},
+    sources::{
+        opentelemetry::config::{LOGS, METRICS, TRACES},
+        util::request_limiter::RequestLimiterPermit,
+    },
 };
 
 #[derive(Clone)]
@@ -41,8 +44,12 @@ pub(super) struct Service {
 impl TraceService for Service {
     async fn export(
         &self,
-        request: Request<ExportTraceServiceRequest>,
+        mut request: Request<ExportTraceServiceRequest>,
     ) -> Result<Response<ExportTraceServiceResponse>, Status> {
+        let permit = request
+            .extensions_mut()
+            .remove::<RequestLimiterPermit>()
+            .expect("request limiter layer must attach a permit");
         let events = if let Some(deserializer) = self.deserializer.as_ref() {
             let raw_bytes = request.get_ref().encode_to_vec();
             let bytes = bytes::Bytes::from(raw_bytes);
@@ -58,7 +65,7 @@ impl TraceService for Service {
                 .flat_map(|v| v.into_event_iter())
                 .collect()
         };
-        self.handle_events(events, TRACES).await?;
+        self.handle_events(events, TRACES, permit).await?;
 
         Ok(Response::new(ExportTraceServiceResponse {
             partial_success: None,
@@ -70,8 +77,12 @@ impl TraceService for Service {
 impl LogsService for Service {
     async fn export(
         &self,
-        request: Request<ExportLogsServiceRequest>,
+        mut request: Request<ExportLogsServiceRequest>,
     ) -> Result<Response<ExportLogsServiceResponse>, Status> {
+        let permit = request
+            .extensions_mut()
+            .remove::<RequestLimiterPermit>()
+            .expect("request limiter layer must attach a permit");
         let events = if let Some(deserializer) = self.deserializer.as_ref() {
             let raw_bytes = request.get_ref().encode_to_vec();
             let bytes = bytes::Bytes::from(raw_bytes);
@@ -87,7 +98,7 @@ impl LogsService for Service {
                 .flat_map(|v| v.into_event_iter(self.log_namespace))
                 .collect()
         };
-        self.handle_events(events, LOGS).await?;
+        self.handle_events(events, LOGS, permit).await?;
 
         Ok(Response::new(ExportLogsServiceResponse {
             partial_success: None,
@@ -99,8 +110,12 @@ impl LogsService for Service {
 impl MetricsService for Service {
     async fn export(
         &self,
-        request: Request<ExportMetricsServiceRequest>,
+        mut request: Request<ExportMetricsServiceRequest>,
     ) -> Result<Response<ExportMetricsServiceResponse>, Status> {
+        let permit = request
+            .extensions_mut()
+            .remove::<RequestLimiterPermit>()
+            .expect("request limiter layer must attach a permit");
         let events = if let Some(deserializer) = self.deserializer.as_ref() {
             let raw_bytes = request.get_ref().encode_to_vec();
             // Major caveat here, the output event will be logs.
@@ -118,7 +133,7 @@ impl MetricsService for Service {
                 .collect()
         };
 
-        self.handle_events(events, METRICS).await?;
+        self.handle_events(events, METRICS, permit).await?;
 
         Ok(Response::new(ExportMetricsServiceResponse {
             partial_success: None,
@@ -131,6 +146,7 @@ impl Service {
         &self,
         mut events: Vec<Event>,
         log_name: &'static str,
+        permit: RequestLimiterPermit,
     ) -> Result<(), Status> {
         // When using OTLP decoding, count individual items within the batch
         // to maintain consistency with other Vector sources
@@ -139,6 +155,7 @@ impl Service {
         } else {
             events.len()
         };
+        permit.decoding_finished(count);
         let byte_size = events.estimated_json_encoded_size_of();
         self.events_received.emit(CountByteSize(count, byte_size));