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

[Jansen-w]

Summary

Prevent unbounded accumulation of decoded OTLP requests when the pipeline is backpressured.

Problem

The source output channel is bounded, but concurrent HTTP and gRPC handlers can decode complete requests before waiting to enqueue their events. When downstream processing slows or stops, those suspended handlers retain their decoded events outside the channel. Consequently, channel capacity does not bound total ingress memory, and increasing client concurrency can exhaust process memory even with a bounded downstream buffer.

Solution

• Apply shared, source-wide admission control before HTTP body collection and gRPC decompression/decoding. Excess requests are rejected immediately rather than queued for admission.
• Reuse the adaptive request limiter already used by TCP sources. It starts with up to two concurrent requests, adapts to observed decoded request sizes, and is capped by max_concurrent_requests. When unset, the maximum uses configured runtime worker count, with available parallelism as a fallback.
• Hold admission permits through decoding, pipeline sends, and end-to-end acknowledgement waits. Completion, cancellation, or timeout releases the permit.
• Return retryable overload responses: HTTP 429 and gRPC UNAVAILABLE. Using UNAVAILABLE avoids the OTLP non-retryable interpretation of RESOURCE_EXHAUSTED without RetryInfo.
• Add request_timeout_secs, defaulting to 30 seconds, so incomplete uploads and stalled requests cannot hold permits indefinitely while awaiting progress. Timeouts return HTTP 503 or gRPC UNAVAILABLE. This timeout is cooperative and cannot preempt synchronous decoding.

This bounds concurrent request processing, not total process memory in bytes. Request sizes, decoding overhead, and downstream buffers still contribute to memory usage. A timeout does not roll back events already enqueued, so retries can duplicate previously accepted events.

Commit structure

1. Move the existing adaptive limiter into shared source utilities, without behavioral changes.
2. Add fail-fast acquisition and support for a configured maximum of one, preserving the existing TCP/framestream minimum.
3. Integrate OTLP admission and timeout controls, regression tests, configuration documentation, and changelog.

Vector configuration

Example of explicit controls:

sources:
  otlp:
    type: opentelemetry
    grpc:
      address: 127.0.0.1:4317
    http:
      address: 127.0.0.1:4318
    max_concurrent_requests: 8
    request_timeout_secs: 30

Existing configurations gain adaptive admission and the default timeout. Both settings must be positive when specified. Operators can tune them for request sizes, memory availability, and expected delivery latency.

How did you test this PR?

Before/after memory reproduction

Both local runs exercised OTLP/HTTP logs with 16 runtime workers, 300 concurrent clients, and 12,000 records per request, with a 3,000-byte message body per record. Each protobuf request was 36,108,010 bytes (about 34.44 MiB). The pipeline had no transforms and used a directly disk-buffered HTTP sink whose endpoint deliberately stopped responding. The disk buffer was filled before the large-request burst.

Memory was measured from native macOS process RSS using ps -o rss, converted from KiB to bytes. These were not Kubernetes/cgroup measurements or OOM-killer tests.

Measurement	Earlier pre-fix reproduction	Patched Vector (08b2de4725)
RSS before large-request burst in the earlier run / idle baseline in the patched run	About 228 MiB	77.7 MiB
Peak observed RSS	13.68 GiB, stopped by watchdog	1.27 GiB, watchdog did not trigger
Growth pattern	Crossed 12 GiB in about 3 seconds	Stayed below the 2 GiB safety cutoff throughout the bounded run
RSS after admitted requests timed out	Not measured: worker terminated at high RSS	829 MiB, stable during the final five-second observation
Disk-buffer files during burst	Flat at approximately 128 MiB	Flat at 125.7 MiB

For the patched run:

• Built native debug Vector with sources-opentelemetry,sinks-http and used default adaptive admission and the 30-second timeout.
• Sent 24 successful small requests first to warm the limiter above its cold-start capacity, then filled the disk buffer. A prefill request returned HTTP 503 after approximately 30 seconds, and disk usage remained unchanged for five seconds before the burst.
• Issued a single burst of 300 clients with no retry loop. Ten clients completed body transmission and received HTTP 503 after approximately 30 seconds while downstream remained blocked.
• The other 290 clients encountered write-side broken pipes or connection resets before finishing their uploads. The harness could not read their early responses, so these are not counted as confirmed HTTP 429 responses. Separate protocol tests verify the 429 response.
• Sampled RSS at a target interval of 100 ms, collecting 636 samples over approximately 72 seconds. All spawned processes were cleaned up.

Interpretation: The patched run did not reproduce the earlier multi-GiB request accumulation, despite the same offered request size and concurrency. This is a behavioral regression check, not an apples-to-apples performance benchmark: the two runs used different builds and harnesses, their baselines differ, and there was no same-build unpatched control. It does not establish a universal RSS ceiling or cover gRPC load at this scale.

Automated checks

• 29 OpenTelemetry tests passed, including shared HTTP/gRPC admission, retryable overload responses, trailing-slash/query-string compatibility, and timeout recovery during acknowledgement waits.
• Five adaptive limiter tests passed, including fail-fast acquisition and a maximum of one.
• gRPC admission, cancellation, pre-decompression rejection, and timeout tests passed.
• HTTP incomplete-body timeout and permit-recovery coverage passed.
• Production-library Clippy, relevant minimal-feature builds, Rust formatting, diff checks, and changelog validation passed.

Local validation limitations: CUE was unavailable for generated-doc validation, Node was unavailable for repository-wide Markdown/Prettier checks, and test-target Clippy encounters a pre-existing unused_async warning in src/test_util/http.rs.

Does this PR include user facing changes?

• Yes. Includes configuration documentation and a security changelog fragment.
• No.