Skip to content

Commit caa705a

Browse files
muhammetsafakmuhammetsafak
authored
feat(otel): optional OpenTelemetry tracing module (ADR-0025) (#2)
Mirrors the Go/Python/Node otel modules: a new BabelQueue\Otel\Tracing (open-telemetry/api is an optional 'suggest' dependency, so the core stays dependency-light) emitting produce/consume spans correlated across hops via trace_id<->32-hex OTel TraceId. Tracing::wrap (consumer span, mirroring SchemaValidated/Idempotent::wrap) + Tracing::publish (producer span). Envelope untouched (GR-1); opt-in. phpunit 160 green; phpstan level 9 clean.
1 parent a465962 commit caa705a

3 files changed

Lines changed: 431 additions & 1 deletion

File tree

‎composer.json‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@
2727
},
2828
"require-dev": {
2929
"mockery/mockery": "^1.6",
30+
"open-telemetry/api": "^1.9",
31+
"open-telemetry/sdk": "^1.14",
3032
"php-amqplib/php-amqplib": "^3.5",
3133
"phpstan/phpstan": "^2.0",
3234
"phpunit/phpunit": "^10.5|^11.0",
@@ -39,7 +41,8 @@
3941
"aws/aws-sdk-php": "For the framework-less Amazon SQS transport (BabelQueue\\Transport\\SqsTransport).",
4042
"stomp-php/stomp-php": "To produce to Apache ActiveMQ Artemis over STOMP (the \u00a77 PHP path) via StompTransport.",
4143
"ext-rdkafka": "To produce to Apache Kafka (the \u00a76 PHP path) via KafkaTransport (the php-rdkafka PECL extension over librdkafka; opt-in, relaxes GR-7 for Kafka \u2014 ADR-0019).",
42-
"textalk/websocket": "Pure-PHP WebSocket client to produce to Apache Pulsar (the \u00a75 PHP path) via PulsarTransport over Pulsar's WebSocket API (GR-7 intact \u2014 ADR-0020)."
44+
"textalk/websocket": "Pure-PHP WebSocket client to produce to Apache Pulsar (the \u00a75 PHP path) via PulsarTransport over Pulsar's WebSocket API (GR-7 intact \u2014 ADR-0020).",
45+
"open-telemetry/api": "To emit OpenTelemetry produce/consume spans via BabelQueue\\Otel\\Tracing (opt-in, ADR-0025; also install open-telemetry/sdk to export)."
4346
},
4447
"autoload": {
4548
"psr-4": {

‎src/Otel/Tracing.php‎

Lines changed: 190 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,190 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace BabelQueue\Otel;
6+
7+
use BabelQueue\Codec\EnvelopeCodec;
8+
use BabelQueue\Contracts\ConsumedMessage;
9+
use BabelQueue\Contracts\Transport;
10+
use OpenTelemetry\API\Trace\Span;
11+
use OpenTelemetry\API\Trace\SpanContext;
12+
use OpenTelemetry\API\Trace\SpanKind;
13+
use OpenTelemetry\API\Trace\StatusCode;
14+
use OpenTelemetry\API\Trace\TraceFlags;
15+
use OpenTelemetry\API\Trace\TracerInterface;
16+
use OpenTelemetry\Context\Context;
17+
use OpenTelemetry\Context\ContextInterface;
18+
use Throwable;
19+
20+
/**
21+
* Optional OpenTelemetry tracing for a babelqueue producer or consumer (ADR-0025) — the PHP
22+
* mirror of the Go `babelqueue-go/otel`, Python `babelqueue.otel` and Node
23+
* `@babelqueue/core/otel` modules.
24+
*
25+
* It emits a CONSUMER span per handled message and a PRODUCER span per publish, correlating
26+
* them across every hop and SDK through the envelope's `trace_id` — a UUID, which maps 1:1 to
27+
* a 32-hex OTel trace id. The wire envelope is untouched (GR-1) and the dependency-light core
28+
* never requires OpenTelemetry: `open-telemetry/api` is an **optional** (`suggest`) dependency
29+
* and this class is only loaded when you wire a tracer.
30+
*
31+
* $dispatch->on('urn:babel:orders:created', Tracing::wrap($tracer, $handler)); // consumer
32+
* Tracing::publish($tracer, $transport, 'urn:babel:orders:created', $data); // producer
33+
*
34+
* Every hop that shares a `trace_id` shares one OTel trace. Exact cross-hop *span*
35+
* parent-child linkage (W3C `traceparent` as a transport header) is a documented follow-up.
36+
*/
37+
final class Tracing
38+
{
39+
private const SYSTEM = 'babelqueue';
40+
private const INVALID_TRACE_ID = '00000000000000000000000000000000';
41+
private const INVALID_SPAN_ID = '0000000000000000';
42+
43+
/**
44+
* Map an envelope `trace_id` to a deterministic 32-hex OTel trace id: a UUID maps to its
45+
* hex bytes; any other string is hashed (SHA-256, first 16 bytes). The inverse of
46+
* {@see self::uuidOf()} for the UUID case.
47+
*/
48+
public static function traceIdOf(string $traceId): string
49+
{
50+
$hex = strtolower(str_replace('-', '', $traceId));
51+
if (preg_match('/^[0-9a-f]{32}$/', $hex) === 1 && $hex !== self::INVALID_TRACE_ID) {
52+
return $hex;
53+
}
54+
55+
return substr(hash('sha256', $traceId), 0, 32);
56+
}
57+
58+
/**
59+
* Format a 32-hex OTel trace id as a canonical UUID string — the form a producer stamps
60+
* into the message's `trace_id` so a consumer can recover the same trace id via
61+
* {@see self::traceIdOf()}.
62+
*/
63+
public static function uuidOf(string $traceIdHex): string
64+
{
65+
$hex = substr(str_pad(strtolower(str_replace('-', '', $traceIdHex)), 32, '0', STR_PAD_LEFT), 0, 32);
66+
67+
return sprintf(
68+
'%s-%s-%s-%s-%s',
69+
substr($hex, 0, 8),
70+
substr($hex, 8, 4),
71+
substr($hex, 12, 4),
72+
substr($hex, 16, 4),
73+
substr($hex, 20, 12),
74+
);
75+
}
76+
77+
/**
78+
* Wrap a consume handler so each message emits a CONSUMER span `process <urn>` in the OTel
79+
* trace derived from its `trace_id`, recording the handler's error/status. Mirrors the
80+
* {@see \BabelQueue\Schema\SchemaValidated::wrap()} / {@see \BabelQueue\Idempotency\Idempotent::wrap()}
81+
* shape and composes with the consume runtime's ack-on-return / redeliver-on-throw contract.
82+
*
83+
* @param callable(ConsumedMessage): void $handler
84+
* @return callable(ConsumedMessage): void
85+
*/
86+
public static function wrap(TracerInterface $tracer, callable $handler): callable
87+
{
88+
return static function (ConsumedMessage $message) use ($tracer, $handler): void {
89+
$meta = $message->getMeta();
90+
$span = $tracer->spanBuilder('process ' . $message->getUrn())
91+
->setParent(self::parentContext($message->getTraceId()))
92+
->setSpanKind(SpanKind::KIND_CONSUMER)
93+
->setAttributes([
94+
'messaging.system' => self::SYSTEM,
95+
'messaging.operation' => 'process',
96+
'messaging.destination.name' => self::stringMeta($meta, 'queue'),
97+
'messaging.message.id' => self::stringMeta($meta, 'id'),
98+
'messaging.message.conversation_id' => $message->getTraceId(),
99+
'messaging.babelqueue.attempts' => $message->attempts(),
100+
])
101+
->startSpan();
102+
$scope = $span->activate();
103+
104+
try {
105+
$handler($message);
106+
} catch (Throwable $e) {
107+
$span->recordException($e);
108+
$span->setStatus(StatusCode::STATUS_ERROR, $e->getMessage());
109+
110+
throw $e;
111+
} finally {
112+
$scope->detach();
113+
$span->end();
114+
}
115+
};
116+
}
117+
118+
/**
119+
* Publish under a PRODUCER span `publish <urn>`, carrying the active trace's id into the
120+
* built envelope's `trace_id` so the downstream consumer recovers the same trace. Encodes
121+
* and publishes through the given {@see Transport}; returns the transport's message id.
122+
*
123+
* @param array<string, mixed> $data
124+
*/
125+
public static function publish(
126+
TracerInterface $tracer,
127+
Transport $transport,
128+
string $urn,
129+
array $data = [],
130+
string $queue = 'default',
131+
): ?string {
132+
$span = $tracer->spanBuilder('publish ' . $urn)
133+
->setSpanKind(SpanKind::KIND_PRODUCER)
134+
->setAttributes([
135+
'messaging.system' => self::SYSTEM,
136+
'messaging.operation' => 'publish',
137+
'messaging.destination.name' => $urn,
138+
])
139+
->startSpan();
140+
$scope = $span->activate();
141+
142+
try {
143+
$traceId = self::uuidOf($span->getContext()->getTraceId());
144+
$envelope = EnvelopeCodec::make($urn, $data, $queue, $traceId);
145+
$result = $transport->publish(EnvelopeCodec::encode($envelope), $queue);
146+
$span->setAttribute('messaging.message.id', self::stringMeta($envelope['meta'], 'id'));
147+
148+
return $result;
149+
} catch (Throwable $e) {
150+
$span->recordException($e);
151+
$span->setStatus(StatusCode::STATUS_ERROR, $e->getMessage());
152+
153+
throw $e;
154+
} finally {
155+
$scope->detach();
156+
$span->end();
157+
}
158+
}
159+
160+
/**
161+
* A context carrying a remote parent in the `trace_id`-derived trace, so a span started
162+
* from it lands in that trace (cross-hop correlation). The parent span id is derived
163+
* deterministically (and non-zero) so the context is valid.
164+
*/
165+
private static function parentContext(string $traceId): ContextInterface
166+
{
167+
$spanContext = SpanContext::createFromRemoteParent(
168+
self::traceIdOf($traceId),
169+
self::spanIdOf($traceId),
170+
TraceFlags::SAMPLED,
171+
);
172+
173+
return Context::getCurrent()->withContextValue(Span::wrap($spanContext));
174+
}
175+
176+
private static function spanIdOf(string $traceId): string
177+
{
178+
$spanId = substr(hash('sha256', 'babelqueue-span:' . $traceId), 0, 16);
179+
180+
return $spanId === self::INVALID_SPAN_ID ? '0000000000000001' : $spanId;
181+
}
182+
183+
/**
184+
* @param array<string, mixed> $meta
185+
*/
186+
private static function stringMeta(array $meta, string $key): string
187+
{
188+
return isset($meta[$key]) && is_string($meta[$key]) ? $meta[$key] : '';
189+
}
190+
}

0 commit comments

Comments
 (0)