25035624412

Committed 28 Apr 2026 05:27AM UTC coverage: 92.409% (+0.7%) from 91.751%

Build # 25035624412

Build Type

push

github

Committed by

web-flow

Commit Message

feat(Instrumentation/Messenger): implement worker and middleware instrumentation (#173)

* chore(messenger): add trace parent in dispatched messages (AMQP)

* chore(messenger): use Messenger events to start/end span for instrumentation

* remove strict type, merge start/end span into 1 subscriber, close span on error and on message handled

* propagation not related to AMQP + read incoming trace (async context)

* rename event subscriber

* chore(messenger): use Messenger events to start/end span for instrumentation

* chore(messenger): add trace parent in dispatched messages (AMQP)

* remove strict type, merge start/end span into 1 subscriber, close span on error and on message handled

* propagation not related to AMQP + read incoming trace (async context)

* rename event subscriber

* fix(messenger): clean up worker subscriber and add functional tests

Fix several issues in the WorkerMessageEventSubscriber introduced by PR #173:
- Replace SDK Span import with API Span to respect API/SDK separation
- Implement InstrumentationTypeInterface for consistency with other subscribers
- Add event priorities (10000/-10000) to wrap all other processing
- Add messaging semantic convention attributes (operation.type, destination.name)
- Include message class name in span name for better trace readability
- Remove stale imports and duplicate propagation middleware service definition
- Clean up propagation middleware when messenger tracing is disabled
- Add PHPStan baseline entries for untyped $carrier interface params
- Add functional tests for worker message handled, failed, and attribute mode
- Reorganize messenger tests into Messenger/ subdirectory
- Disable retry on test transport to isolate worker span assertions

* test(messenger): add tests for transport tracing and propagation middleware

Cover TraceableMessengerTransport (get/ack/reject spans + TransportException
error recording) and AddStampForPropagationMiddleware (stamp skip, no-scope
passthrough, ac... (continued)

Coverage Stats

187 of 191 new or added lines in 14 files covered. (97.91%)

12 existing lines in 10 files now uncovered.

2459 of 2661 relevant lines covered (92.41%)

15.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.5

/src/DependencyInjection/OpenTelemetryLogsExtension.php

<?php

namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\DependencyInjection;

use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Exporter\ExporterOptionsInterface;
use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Log\LogProcessor\LogProcessorEnum;
use Monolog\Level;
use OpenTelemetry\API\Common\Time\ClockInterface;
use OpenTelemetry\API\Logs\LoggerInterface;
use OpenTelemetry\Contrib\Logs\Monolog\Handler as MonologHandler;
use Symfony\Component\DependencyInjection\ChildDefinition;
use Symfony\Component\DependencyInjection\ContainerBuilder;
use Symfony\Component\DependencyInjection\Definition;
use Symfony\Component\DependencyInjection\Reference;

/**
 * @phpstan-import-type ExporterOptions from ExporterOptionsInterface
 */
final class OpenTelemetryLogsExtension
{
    /**
     * @var array<string, mixed>
     */
    private array $config;
    private ContainerBuilder $container;

    /**
     * @param array{
     *     monolog: array<string, mixed>,
     *     loggers: array<string, mixed>,
     *     exporters: array<string, mixed>,
     *     processors: array<string, mixed>,
     *     providers: array<string, mixed>
     * }|array<string, mixed> $config
     */
    public function __invoke(array $config, ContainerBuilder $container): void
    {
        $this->config = $config;
        $this->container = $container;

        foreach ($this->config['exporters'] as $name => $exporter) {
            $this->loadLogExporter($name, $exporter);
        }

        foreach ($this->config['processors'] as $name => $processor) {
            $this->loadLogProcessor($name, $processor);
        }

        foreach ($this->config['providers'] as $name => $provider) {
            $this->loadLogProvider($name, $provider);
        }

        foreach ($this->config['loggers'] as $name => $logger) {
            $this->loadLogLogger($name, $logger);
        }

        $defaultLogger = null;
        if (0 < count($this->config['loggers'])) {
            $defaultLogger = array_key_first($this->config['loggers']);
        }

        if (null !== $defaultLogger) {
            $this->container->setAlias('open_telemetry.logs.default_logger', new Reference(sprintf('open_telemetry.logs.loggers.%s', $defaultLogger)));
            $this->container->setAlias(LoggerInterface::class, new Reference(sprintf('open_telemetry.logs.loggers.%s', $defaultLogger)));
        }

        $this->loadMonologHandlers();
    }

    /**
     * @param array{
     *      dsn: string,
     *      options?: ExporterOptions
     *  } $config
     */
    private function loadLogExporter(string $name, array $config): void
    {
        $dsn = (new ChildDefinition('open_telemetry.exporter_dsn'))->setArguments([$config['dsn']]);
        $exporterOptions = (new ChildDefinition('open_telemetry.otlp_exporter_options'))->setArguments([$config['options'] ?? []]);

        $this->container
            ->setDefinition(
                sprintf('open_telemetry.logs.exporters.%s', $name),
                new ChildDefinition('open_telemetry.logs.exporter_interface'),
            )
            ->setArguments([$dsn, $exporterOptions]);
    }

    /**
     * @param array{
     *      type: string,
     *      processors?: string[],
     *      batch?: array{
     *          clock: class-string<ClockInterface>,
     *          max_queue_size: int,
     *          schedule_delay: int,
     *          export_timeout: int,
     *          max_export_batch_size: int,
     *          auto_flush: bool,
     *          meter_provider?: string,
     *      },
     *      exporter?: string
     *  } $config
     */
    private function loadLogProcessor(string $name, array $config): void
    {
        $this->container
            ->setDefinition(
                sprintf('open_telemetry.logs.processors.%s', $name),
                new ChildDefinition('open_telemetry.logs.processor_interface')
            )
            ->setFactory([new Reference(sprintf('open_telemetry.logs.processor_factory.%s', $config['type'])), 'createProcessor'])
            ->setArguments([
                array_map(fn (string $processor) => new Reference($processor), $config['processors'] ?? []),
                LogProcessorEnum::Batch->value === $config['type'] && isset($config['batch']) ? [
                    'clock' => new Reference($config['batch']['clock']),
                    'max_queue_size' => $config['batch']['max_queue_size'],
                    'schedule_delay' => $config['batch']['schedule_delay'],
                    'export_timeout' => $config['batch']['export_timeout'],
                    'max_export_batch_size' => $config['batch']['max_export_batch_size'],
                    'auto_flush' => $config['batch']['auto_flush'],
                    'meter_provider' => isset($config['batch']['meter_provider']) ? new Reference($config['batch']['meter_provider']) : null,
                ] : null,
                isset($config['exporter']) ? new Reference($config['exporter']) : null,
            ]);
    }

    /**
     * @param array{
     *     type: string,
     *     processor?: string,
     * } $config
     */
    private function loadLogProvider(string $name, array $config): void
    {
        $this->container
            ->setDefinition(
                sprintf('open_telemetry.logs.providers.%s', $name),
                new ChildDefinition('open_telemetry.logs.provider_interface')
            )
            ->setFactory([new Reference(sprintf('open_telemetry.logs.provider_factory.%s', $config['type'])), 'createProvider'])
            ->setArguments([
                isset($config['processor']) ? new Reference($config['processor']) : null,
                new Reference('open_telemetry.resource_info'),
            ]);
    }

    /**
     * @param array{
     *     provider: string,
     *     name?: string,
     *     version?: string,
     * } $config
     */
    private function loadLogLogger(string $name, array $config): void
    {
        $this->container
            ->setDefinition(
                sprintf('open_telemetry.logs.loggers.%s', $name),
                new ChildDefinition('open_telemetry.logs.logger_interface'),
            )
            ->setPublic(true)
            ->setFactory([new Reference($config['provider']), 'getLogger'])
            ->setArguments([
                $config['name'] ?? $this->container->getParameter('open_telemetry.bundle.name'),
                $config['version'] ?? $this->container->getParameter('open_telemetry.bundle.version'),
            ]);
    }

    private function loadMonologHandlers(): void
    {
        if (false === $this->config['monolog']['enabled']) {
            return;
        }

        if (!class_exists(MonologHandler::class)) {
            throw new \LogicException('To configure the Monolog handler, you must first install the open-telemetry/opentelemetry-logger-monolog package.');
        }

        $this->container->setDefinition('open_telemetry.logs.monolog.handler', new Definition(MonologHandler::class));

        foreach ($this->config['monolog']['handlers'] as $name => $handler) {
            $handlerId = sprintf('open_telemetry.logs.monolog.handlers.%s', $name);
            $this->container
                ->setDefinition($handlerId, new ChildDefinition('open_telemetry.logs.monolog.handler'))
                ->setPublic(true)
                ->setArguments([
                    '$loggerProvider' => new Reference($handler['provider']),
                    '$level' => Level::fromName(ucfirst($handler['level'])),
                    '$bubble' => $handler['bubble'],
                ]);
        }
    }
}

1	<?php
2
3	namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\DependencyInjection;
4
5	use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Exporter\ExporterOptionsInterface;
6	use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Log\LogProcessor\LogProcessorEnum;
7	use Monolog\Level;
8	use OpenTelemetry\API\Common\Time\ClockInterface;
9	use OpenTelemetry\API\Logs\LoggerInterface;
10	use OpenTelemetry\Contrib\Logs\Monolog\Handler as MonologHandler;
11	use Symfony\Component\DependencyInjection\ChildDefinition;
12	use Symfony\Component\DependencyInjection\ContainerBuilder;
13	use Symfony\Component\DependencyInjection\Definition;
14	use Symfony\Component\DependencyInjection\Reference;
15
16	/**
17	* @phpstan-import-type ExporterOptions from ExporterOptionsInterface
18	*/
19	final class OpenTelemetryLogsExtension
20	{
21	/**
22	* @var array<string, mixed>
23	*/
24	private array $config;
25	private ContainerBuilder $container;
26
27	/**
28	* @param array{
29	* monolog: array<string, mixed>,
30	* loggers: array<string, mixed>,
31	* exporters: array<string, mixed>,
32	* processors: array<string, mixed>,
33	* providers: array<string, mixed>
34	* }\|array<string, mixed> $config
35	*/
36	public function __invoke(array $config, ContainerBuilder $container): void
37	{
38	$this->config = $config;	34✔
39	$this->container = $container;	34✔
40
41	foreach ($this->config['exporters'] as $name => $exporter) {	34✔
42	$this->loadLogExporter($name, $exporter);	12✔
43	}
44
45	foreach ($this->config['processors'] as $name => $processor) {	34✔
46	$this->loadLogProcessor($name, $processor);	14✔
47	}
48
49	foreach ($this->config['providers'] as $name => $provider) {	34✔
50	$this->loadLogProvider($name, $provider);	14✔
51	}
52
53	foreach ($this->config['loggers'] as $name => $logger) {	34✔
54	$this->loadLogLogger($name, $logger);	14✔
55	}
56
57	$defaultLogger = null;	34✔
58	if (0 < count($this->config['loggers'])) {	34✔
59	$defaultLogger = array_key_first($this->config['loggers']);	14✔
60	}
61
62	if (null !== $defaultLogger) {	34✔
63	$this->container->setAlias('open_telemetry.logs.default_logger', new Reference(sprintf('open_telemetry.logs.loggers.%s', $defaultLogger)));	14✔
64	$this->container->setAlias(LoggerInterface::class, new Reference(sprintf('open_telemetry.logs.loggers.%s', $defaultLogger)));	14✔
65	}
66
67	$this->loadMonologHandlers();	34✔
68	}
69
70	/**
71	* @param array{
72	* dsn: string,
73	* options?: ExporterOptions
74	* } $config
75	*/
76	private function loadLogExporter(string $name, array $config): void
77	{
78	$dsn = (new ChildDefinition('open_telemetry.exporter_dsn'))->setArguments([$config['dsn']]);	12✔
79	$exporterOptions = (new ChildDefinition('open_telemetry.otlp_exporter_options'))->setArguments([$config['options'] ?? []]);	12✔
80
81	$this->container	12✔
82	->setDefinition(	12✔
83	sprintf('open_telemetry.logs.exporters.%s', $name),	12✔
84	new ChildDefinition('open_telemetry.logs.exporter_interface'),	12✔
85	)	12✔
86	->setArguments([$dsn, $exporterOptions]);	12✔
87	}
88
89	/**
90	* @param array{
91	* type: string,
92	* processors?: string[],
93	* batch?: array{
94	* clock: class-string<ClockInterface>,
95	* max_queue_size: int,
96	* schedule_delay: int,
97	* export_timeout: int,
98	* max_export_batch_size: int,
99	* auto_flush: bool,
100	* meter_provider?: string,
101	* },
102	* exporter?: string
103	* } $config
104	*/
105	private function loadLogProcessor(string $name, array $config): void
106	{
107	$this->container	14✔
108	->setDefinition(	14✔
109	sprintf('open_telemetry.logs.processors.%s', $name),	14✔
110	new ChildDefinition('open_telemetry.logs.processor_interface')	14✔
111	)	14✔
112	->setFactory([new Reference(sprintf('open_telemetry.logs.processor_factory.%s', $config['type'])), 'createProcessor'])	14✔
113	->setArguments([	14✔
114	array_map(fn (string $processor) => new Reference($processor), $config['processors'] ?? []),	14✔
115	LogProcessorEnum::Batch->value === $config['type'] && isset($config['batch']) ? [	14✔
116	'clock' => new Reference($config['batch']['clock']),	14✔
117	'max_queue_size' => $config['batch']['max_queue_size'],	14✔
118	'schedule_delay' => $config['batch']['schedule_delay'],	14✔
119	'export_timeout' => $config['batch']['export_timeout'],	14✔
120	'max_export_batch_size' => $config['batch']['max_export_batch_size'],	14✔
121	'auto_flush' => $config['batch']['auto_flush'],	14✔
122	'meter_provider' => isset($config['batch']['meter_provider']) ? new Reference($config['batch']['meter_provider']) : null,	14✔
UNCOV 123	] : null,	×
124	isset($config['exporter']) ? new Reference($config['exporter']) : null,	14✔
125	]);	14✔
126	}
127
128	/**
129	* @param array{
130	* type: string,
131	* processor?: string,
132	* } $config
133	*/
134	private function loadLogProvider(string $name, array $config): void
135	{
136	$this->container	14✔
137	->setDefinition(	14✔
138	sprintf('open_telemetry.logs.providers.%s', $name),	14✔
139	new ChildDefinition('open_telemetry.logs.provider_interface')	14✔
140	)	14✔
141	->setFactory([new Reference(sprintf('open_telemetry.logs.provider_factory.%s', $config['type'])), 'createProvider'])	14✔
142	->setArguments([	14✔
143	isset($config['processor']) ? new Reference($config['processor']) : null,	14✔
144	new Reference('open_telemetry.resource_info'),	14✔
145	]);	14✔
146	}
147
148	/**
149	* @param array{
150	* provider: string,
151	* name?: string,
152	* version?: string,
153	* } $config
154	*/
155	private function loadLogLogger(string $name, array $config): void
156	{
157	$this->container	14✔
158	->setDefinition(	14✔
159	sprintf('open_telemetry.logs.loggers.%s', $name),	14✔
160	new ChildDefinition('open_telemetry.logs.logger_interface'),	14✔
161	)	14✔
162	->setPublic(true)	14✔
163	->setFactory([new Reference($config['provider']), 'getLogger'])	14✔
164	->setArguments([	14✔
165	$config['name'] ?? $this->container->getParameter('open_telemetry.bundle.name'),	14✔
166	$config['version'] ?? $this->container->getParameter('open_telemetry.bundle.version'),	14✔
167	]);	14✔
168	}
169
170	private function loadMonologHandlers(): void
171	{
172	if (false === $this->config['monolog']['enabled']) {	34✔
173	return;	25✔
174	}
175
176	if (!class_exists(MonologHandler::class)) {	9✔
177	throw new \LogicException('To configure the Monolog handler, you must first install the open-telemetry/opentelemetry-logger-monolog package.');	×
178	}
179
180	$this->container->setDefinition('open_telemetry.logs.monolog.handler', new Definition(MonologHandler::class));	9✔
181
182	foreach ($this->config['monolog']['handlers'] as $name => $handler) {	9✔
183	$handlerId = sprintf('open_telemetry.logs.monolog.handlers.%s', $name);	9✔
184	$this->container	9✔
185	->setDefinition($handlerId, new ChildDefinition('open_telemetry.logs.monolog.handler'))	9✔
186	->setPublic(true)	9✔
187	->setArguments([	9✔
188	'$loggerProvider' => new Reference($handler['provider']),	9✔
189	'$level' => Level::fromName(ucfirst($handler['level'])),	9✔
190	'$bubble' => $handler['bubble'],	9✔
191	]);	9✔
192	}
193	}
194	}

FriendsOfOpenTelemetry / opentelemetry-bundle / 25035624412

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous