Skip to content

Commit a55b612

Browse files
authored
refactor(flow-php/telemetry-otlp-bridge): split sync and async curl transports (#2482)
* refactor(flow-php/telemetry-otlp-bridge): split sync and async curl transports - CurlTransport now synchronous: send() blocks and throws on failure - add AsyncCurlTransport (curl_multi) with bounded select-driven tick() - otlp_async_curl_transport()/otlp_async_curl_options() DSL helpers - bundle: transport.type 'async_curl', auto-pumped on WorkerRunningEvent - raise curl request timeout to 10000ms; async connect 1500ms, pump 100ms - fix CollectorMetrics parser for scientific-notation counters * fix: update dsl definitons
1 parent 5077649 commit a55b612

21 files changed

Lines changed: 2221 additions & 655 deletions

File tree

documentation/components/bridges/symfony-telemetry-bundle.md

Lines changed: 33 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -813,24 +813,36 @@ exporters:
813813

814814
Inside `exporters.<name>.otlp.transport`. Required for the `otlp` sub-block.
815815

816+
#### Transport type
817+
818+
| `type` | Transport | Notes |
819+
|--------------|----------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------|
820+
| `curl` | `CurlTransport` | Synchronous HTTP (default). Each `send()` blocks until the response. |
821+
| `async_curl` | `AsyncCurlTransport` | Non-blocking `curl_multi`. Auto-pumped on Messenger's `WorkerRunningEvent` when `messenger` instrumentation is enabled; otherwise pump `tick()` yourself. |
822+
| `grpc` | `GrpcTransport` | OTLP/gRPC (Protobuf only). |
823+
| `stream` | `StreamTransport` | JSONL to a file path or `php://` stream. |
824+
| `service` | user service | Aliases an existing transport service id. |
825+
826+
Prefer `curl`. Use `async_curl` only for non-blocking dispatch — in a Messenger worker the bundle pumps it on
827+
`WorkerRunningEvent`; elsewhere you must call `tick()` yourself. See the
828+
[OTLP bridge async transport section](/documentation/components/bridges/telemetry-otlp-bridge.md#async-curl-transport).
829+
816830
#### Timeouts
817831

818832
The defaults assume the recommended deployment: an OpenTelemetry Collector running close to the application (loopback,
819833
UDS, or sidecar):
820834

821-
| Setting | Default | Applies to | Bounds |
822-
|-----------------------|---------------------:|------------|-------------------------------------------------------------|
823-
| `timeout_ms` | 5000 curl / 250 grpc | curl, grpc | Per-request deadline (curl: total request; grpc: per-call) |
824-
| `connect_timeout_ms` | 250 | curl only | TCP/TLS connect; gRPC has no separate bound |
825-
| `shutdown_timeout_ms` | 5000 | curl, grpc | Wall-clock budget for draining pending requests at shutdown |
835+
| Setting | Default | Applies to | Bounds |
836+
|-----------------------|----------------------------------------:|------------------------|-------------------------------------------------------------|
837+
| `timeout_ms` | 10000 curl / 5000 async_curl / 250 grpc | curl, async_curl, grpc | Per-request deadline (curl: total request; grpc: per-call) |
838+
| `connect_timeout_ms` | 250 curl / 1500 async_curl | curl, async_curl | TCP/TLS connect; gRPC has no separate bound |
839+
| `pump_timeout_ms` | 100 | async_curl only | Per-`tick()` bounded drive budget (`0` = single exec round) |
840+
| `shutdown_timeout_ms` | 5000 | curl, async_curl, grpc | Wall-clock budget for draining pending requests at shutdown |
826841

827-
The curl transport is asynchronous and only advances while the host pumps it, so its `timeout_ms` must span the gap
828-
between dispatch and the next pump — hence the **5000 ms** curl default, which comfortably covers a worker's loop
829-
iteration (the bundle pumps in-flight curl requests on Messenger's `WorkerRunningEvent`, see
830-
[Messenger instrumentation](#messenger)). gRPC calls progress in the background via the grpc core, so the gRPC
831-
per-call deadline stays tight at **250 ms**. `shutdown_timeout_ms` is independent of `timeout_ms` and bounds graceful
832-
drain at exit. For a remote collector across regions, raise `timeout_ms`. See the
833-
[OTLP bridge Timeouts section](/documentation/components/bridges/telemetry-otlp-bridge.md#timeouts) for the rationale.
842+
`curl` is synchronous, so its `timeout_ms` is a per-flush ceiling. `async_curl` uses a larger **1500 ms**
843+
`connect_timeout_ms` (it only advances when pumped) and a **100 ms** `pump_timeout_ms` per `tick()`. gRPC progresses in
844+
the background, so its per-call deadline stays at **250 ms**. `shutdown_timeout_ms` bounds graceful drain at exit. See
845+
the [OTLP bridge Timeouts section](/documentation/components/bridges/telemetry-otlp-bridge.md#timeouts).
834846

835847
#### Failover Transport
836848

@@ -856,8 +868,8 @@ exporters:
856868

857869
- The `failover:` block accepts the same fields as the parent transport, except it cannot itself declare a nested
858870
`failover:` (single-level depth).
859-
- Allowed only on `curl` and `grpc` primaries. `failover` under a `stream` or `service` primary is rejected at
860-
config-validation time.
871+
- Allowed only on `curl`, `async_curl` and `grpc` primaries. `failover` under a `stream` or `service` primary is
872+
rejected at config-validation time.
861873
- The bundle registers `flow.telemetry.exporter.<name>.failover.transport` for the failover service id.
862874

863875
For the underlying behavior — when a forwarded batch is treated as absorbed vs. lost, the shape of
@@ -1156,12 +1168,13 @@ Flushing per message means one exporter round-trip per message. For high-through
11561168
[`max_batch_age`](#batching) on the batching processor so the batch coalesces across messages and a single
11571169
idle worker still exports on a time bound rather than per message.
11581170

1159-
The async OTLP `curl` transport makes no network progress unless the process pumps it, and a worker blocks on its
1160-
queue poll between messages. The bundle therefore pumps every configured `curl` transport on Messenger's
1161-
`WorkerRunningEvent` (each loop iteration), so a request dispatched by the per-message flush completes in the
1162-
background instead of stalling until shutdown and tripping its wall-clock `timeout_ms`. This is wired automatically
1163-
when `messenger` instrumentation is enabled; it requires no configuration. Keep the curl `timeout_ms` comfortably
1164-
above the worker's poll/sleep cadence — the **5000 ms** default already does (see [Timeouts](#timeouts)).
1171+
The OTLP `curl` transport sends synchronously, so the per-message flush exports the batch and reports its outcome
1172+
before the handler returns — there is nothing to pump between messages and no request left in flight to stall until
1173+
shutdown. Each flush blocks up to the curl `timeout_ms`, so keep a Collector close to the worker (loopback/UDS/sidecar)
1174+
to keep that sub-millisecond (see [Timeouts](#timeouts)).
1175+
1176+
With the `async_curl` transport, the subscriber also pumps each transport's `tick()` on every `WorkerRunningEvent`, so
1177+
in-flight requests complete in the background.
11651178

11661179
#### Twig
11671180

documentation/components/bridges/telemetry-otlp-bridge.md

Lines changed: 49 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -155,34 +155,25 @@ The Protobuf serializer requires the `google/protobuf` package.
155155
The production recommendation is to run an OpenTelemetry Collector close to the application (loopback, UDS, or
156156
sidecar), so the roundtrip is sub-millisecond and a stuck collector never freezes your PHP process at shutdown.
157157

158-
The curl transport is asynchronous (`curl_multi`): it only makes network progress while the host process pumps it
159-
(`send()`, `shutdown()`, or `tick()`). Its per-request `timeout_ms` is wall-clock from dispatch, so the budget must
160-
span the gap between dispatching a request and the next pump. In a long-running worker that gap is a whole loop
161-
iteration, which is why curl `timeout_ms` defaults to **5000 ms**. gRPC calls progress in the background via the grpc
162-
core with no host pumping, so the gRPC per-call deadline stays tight at **250 ms**.
163-
164-
| Transport | Setting | Default | Unit | Bounds |
165-
|-----------|-------------------------|--------:|--------------|-------------------------------------------------------------|
166-
| Curl | `withTimeout()` | 5000 | milliseconds | Per-request: connect + send + receive |
167-
| Curl | `withConnectTimeout()` | 250 | milliseconds | TCP/TLS connection establishment only |
168-
| Curl | `withShutdownTimeout()` | 5000 | milliseconds | Wall-clock budget for draining pending requests at shutdown |
169-
| gRPC | `timeoutMs` | 250 | milliseconds | Per-call deadline (no separate connect bound) |
170-
| gRPC | `shutdownTimeoutMs` | 5000 | milliseconds | Wall-clock budget for draining pending calls at shutdown |
171-
172-
`timeout_ms` is the per-request deadline. `shutdown_timeout_ms` is a separate wall-clock budget enforced only when
173-
draining pending requests during `shutdown()` — it bounds graceful exit independently of `timeout_ms`. Pending
174-
requests still in flight after the shutdown deadline are abandoned and reported as failed: forwarded to the failover
175-
transport if one is configured, otherwise surfaced to the curl transport's `ErrorHandler` (default `ErrorLogHandler`).
176-
177-
Failed exports are surfaced to that `ErrorHandler` **as they are reaped** — on `send()`, `tick()`, or `shutdown()`
178-
and never retained, so a long-running host does not accumulate failures. Pass a custom handler as the fifth
179-
`CurlTransport` constructor argument (or the `errorHandler:` argument of `otlp_curl_transport()`); the Symfony bundle
180-
injects the exporter's configured `error_handler` automatically.
181-
182-
For a remote collector across regions or a managed SaaS endpoint, 5000–10000 ms is reasonable for both transports.
183-
Long-running hosts that idle between dispatches (e.g. a Symfony Messenger worker) should call `tick()` periodically so
184-
in-flight curl requests complete in the background instead of stalling until shutdown — see
185-
[Long-running workers](#long-running-workers-tick).
158+
The curl transport is **synchronous**: each `send()` blocks up to `timeout_ms`, then returns or throws. Keeping export
159+
off the hot path is the batching processor's job — it flushes only every `batch_size` signals (or on age / flush /
160+
shutdown). gRPC progresses in the background, so its per-call deadline stays at **250 ms**.
161+
162+
| Transport | Setting | Default | Unit | Bounds |
163+
|-------------|-------------------------|--------:|--------------|------------------------------------------------------------------|
164+
| Curl (sync) | `withTimeout()` | 10000 | milliseconds | Per-request: connect + send + receive (max time `send()` blocks) |
165+
| Curl (sync) | `withConnectTimeout()` | 250 | milliseconds | TCP/TLS connection establishment only |
166+
| Curl (sync) | `withShutdownTimeout()` | 5000 | milliseconds | Reserved for the failover drain budget at shutdown |
167+
| Async curl | `withTimeout()` | 5000 | milliseconds | Per-request wall-clock; must span the gap between pumps |
168+
| Async curl | `withConnectTimeout()` | 1500 | milliseconds | TCP/TLS connect; larger default tolerates infrequent pumping |
169+
| Async curl | `withPumpTimeout()` | 100 | milliseconds | Per-`tick()` bounded drive budget (`0` = single exec round) |
170+
| Async curl | `withShutdownTimeout()` | 5000 | milliseconds | Wall-clock budget for draining pending requests at shutdown |
171+
| gRPC | `timeoutMs` | 250 | milliseconds | Per-call deadline (no separate connect bound) |
172+
| gRPC | `shutdownTimeoutMs` | 5000 | milliseconds | Wall-clock budget for draining pending calls at shutdown |
173+
174+
Against a local collector each send is sub-millisecond, so the defaults are only ceilings. On failure `send()` throws
175+
synchronously — `TransportException`, or `FailoverTransportException` once the batch is forwarded to the failover. Async
176+
curl is tuned differently — see [Asynchronous transport](#async-curl-transport).
186177

187178
```php
188179
<?php
@@ -212,27 +203,42 @@ $remoteGrpc = otlp_grpc_transport(
212203
> earlier versions could not express tight, realistic deadlines. A negative value to `withTimeout()` /
213204
> `withConnectTimeout()` raises `\InvalidArgumentException`.
214205
215-
## Long-running workers (`tick()`) {#long-running-workers-tick}
206+
## Long-running workers {#long-running-workers}
216207

217-
`curl_multi` has no background thread — an in-flight request only advances while the host pumps the multi-handle.
218-
Short-lived processes (an HTTP request, a one-shot CLI command) are fine: they end right after dispatch and
219-
`shutdown()` drains synchronously. A long-running worker is not: it dispatches telemetry after handling a message and
220-
then blocks on its queue poll waiting for the next one. During that idle gap nothing pumps the request, yet
221-
`CURLOPT_TIMEOUT_MS` keeps counting wall-clock — so the request eventually dies with `curl error 28 (Timeout was
222-
reached)` and the telemetry is lost.
208+
The synchronous curl transport completes each request before `send()` returns, so nothing ages out between messages.
209+
Keeping export off the worker's hot path is the **batching processor** — it flushes per `batch_size` signals or age
210+
limit. In a Symfony Messenger worker the bundle flushes after each message and on stop (see the
211+
[Symfony telemetry bundle](/documentation/components/bridges/symfony-telemetry-bundle.md)); keep a local Collector so
212+
each flush stays sub-millisecond. For non-blocking dispatch, see [Asynchronous transport](#async-curl-transport).
223213

224-
`CurlTransport::tick()` fixes this. It is non-blocking: it advances all currently-possible I/O and reaps finished
225-
handles, but never waits. Call it periodically from the host's idle loop so dispatched requests complete in the
226-
background:
214+
## Asynchronous transport (`AsyncCurlTransport`) {#async-curl-transport}
215+
216+
`AsyncCurlTransport` uses `curl_multi` for non-blocking I/O: `send()` queues the request and returns without waiting.
217+
Opt in with `otlp_async_curl_transport()` or the bundle's `transport.type: 'async_curl'`.
218+
219+
A queued request only advances while the host pumps the handle — on the next `send()`, `shutdown()`, or `tick()`.
220+
`tick()` is bounded and select-driven: it drives pending requests for up to `pump_timeout_ms` (default **100 ms**), so a
221+
local backend completes within a single tick. `0` falls back to one non-blocking exec round (rarely enough — prefer the
222+
default).
227223

228224
```php
229-
// once per worker loop iteration
230-
$curlTransport->tick();
225+
use function Flow\Bridge\Telemetry\OTLP\DSL\otlp_async_curl_transport;
226+
227+
$transport = otlp_async_curl_transport('http://localhost:4318');
228+
229+
// Pump once per worker loop so in-flight requests complete:
230+
$transport->tick();
231231
```
232232

233-
`tick()` is a no-op when nothing is pending or after `shutdown()`, so calling it on every iteration is cheap. The
234-
Symfony bundle wires this automatically on Messenger's `WorkerRunningEvent` — see the
235-
[Symfony telemetry bundle](/documentation/components/bridges/symfony-telemetry-bundle.md).
233+
In a Symfony Messenger worker with `messenger` instrumentation enabled, the bundle pumps every `async_curl` transport on
234+
`WorkerRunningEvent` (~1s when idle) — the cadence the **1500 ms** `connect_timeout_ms` default is sized for. Elsewhere
235+
you must call `tick()` yourself.
236+
237+
Failures surface on a later `send()`/`tick()`/`shutdown()` (no caller on the stack), so they go to the failover
238+
transport or, without one, the injected `ErrorHandler` (5th constructor arg, default `ErrorLogHandler`) — which is why
239+
the async transport takes an error handler and the synchronous one just throws.
240+
241+
Prefer the synchronous `curl` transport unless you need non-blocking dispatch and can guarantee a pump cadence.
236242

237243
## Failover Transport
238244

0 commit comments

Comments
 (0)