feat: Explore E2E pipeline latency measurement via synthetic probes (approach 1)

[cijothomas]

Context

Exploring how to measure end-to-end pipeline latency — the time from when a record enters the pipeline to when it is received by the backend.

This PR explores a lightweight approach that does not require any changes to the df-engine itself. The trade-off is that it requires backend cooperation — the backend must recognize probe records and compute the latency. An engine-native approach (where the engine itself tracks and reports latency) would be more self-contained but is likely more complex and expensive to implement correctly. Both directions are worth exploring; this PR starts with the simpler one to validate the concept.

Approach: Synthetic probe traffic + backend-side latency computation

1. A synthetic log record is periodically injected into the pipeline with body = "_OTAP_PROBE" and an _otap_internal.probe.emitted_at_unix_nanos attribute carrying the wall-clock timestamp at emission.

2. Probes flow through the entire real pipeline path as ordinary data — same processors, same exporter batching, same wire format. No special handling anywhere in the engine.

3. The backend detects probes (check if body == "_OTAP_PROBE"), computes now() - emitted_at, emits a latency metric, and drops the record.

Probe generation options

The probe traffic can be generated by:

• probe_emitter receiver (included in this PR) — a minimal in-engine receiver that emits one probe log per configurable interval.
• traffic_generator receiver — the existing traffic generator could serve this purpose once it gains the ability to stamp an origin timestamp on generated telemetry.
• An external script or sidecar — any process that sends an OTLP log with the expected body and attribute to the pipeline's receiver endpoint. No in-engine component needed at all.

Backend convention

Any backend that follows the _OTAP_PROBE convention can participate:

1. On ingest, check if body == "_OTAP_PROBE" (O(1) string check; short-circuit on first byte _)
2. If match: read _otap_internal.probe.emitted_at_unix_nanos, compute now() - emitted_at, emit a latency metric
3. Drop the record (don't index it)

Why backend-side?

Any measurement taken inside the engine structurally cannot include exporter time (batching, serialization, compression, network, ack) — which is typically the dominant contributor to E2E latency. The backend is the only observer that sees the full path.

Demo

The POC uses the Azure Monitor exporter + mock LA server as an example backend, but the same convention would work with any backend that implements the _OTAP_PROBE detection.

To try locally

# Terminal 1 — mock backend (uses mock LA server as example)
cargo run --example mock_la_server -p otap-df-contrib-nodes \
    --features azure-monitor-exporter -- --port 9999

# Terminal 2 — pipeline with probe emitter
cargo run --features azure-monitor-exporter -- \
    --config e2e-latency-probe-poc.yaml --num-cores 1

Output from the mock backend:

🏓 PROBE E2E latency: 450.3ms | source=e2e-latency-poc | id=6e8cc3d2-...
🏓 PROBE E2E latency: 1451.8ms | source=e2e-latency-poc | id=d29009a7-...
🏓 PROBE E2E latency: 2448.2ms | source=e2e-latency-poc | id=feedf721-...

In steady state, probe latency reflects the exporter's batch flush interval. With the default 3s flush, probes show ~450ms–2450ms depending on position in the batch. With a 1s flush, all probes converge to ~450ms. This confirms the measurement captures real pipeline behavior.

What's included

File	Purpose
crates/core-nodes/src/receivers/probe_emitter_receiver/	Probe emitter receiver (config, metrics, mod)
crates/core-nodes/src/processors/probe_sink_processor/	Engine-internal probe sink — detects probes and computes latency within the engine (i.e., up to the processor stage, excluding exporter/network time). Included for comparison but not the recommended approach since it misses the dominant latency contributor (exporter + network).
e2e-latency-probe-poc.yaml	Ready-to-run config: probe_emitter → azure_monitor_exporter → localhost:9999
crates/contrib-nodes/examples/mock_la_server.rs	Mock backend with probe detection (~40 lines added)

Trade-offs

Pros:

• Zero engine changes — probes are just ordinary log records
• Measures the full real path including exporter batching, network, and backend ingestion
• Probe generation is flexible: in-engine receiver, existing traffic generator, or external script

Cons:

• Requires backends to implement the _OTAP_PROBE convention
• Does not give per-stage breakdown within the engine (existing per-node flow metrics already cover that)
• Probes can be dropped by content-based filters if filter rules happen to match the probe body

[lquerel]

I like this approach, and I think it's a good first step. One small suggestion would be to call it backend-observed E2E latency (or simply backend E2E latency) to distinguish it from latency measured by the engine itself.

One limitation is that the measured latency also includes network and backend ingestion time, which may or may not be desirable depending on the use case. This approach also depends on the probe following the same path as the telemetry of interest. For example, a pipeline that filters logs while exporting only metrics may prevent the probe from reaching the backend.

Looking ahead, I wonder whether OTAP PData headers could provide a complementary mechanism for measuring true engine E2E latency. A sampled timestamp attached to the batch header at ingress could be propagated through the engine, with special handling in split, merge, and batching processors. This would avoid introducing synthetic telemetry while keeping the measurement focused on the engine itself.