Laravel: tracing Queues (+ hook refactoring)

src/Instrumentation/Laravel/.php-cs-fixer.php

@@ -40,4 +40,3 @@
     ])
     ->setRiskyAllowed(true)
     ->setFinder($finder);
-

src/Instrumentation/Laravel/composer.json

@@ -7,18 +7,18 @@
   "readme": "./README.md",
   "license": "Apache-2.0",
   "minimum-stability": "dev",
+  "prefer-stable": true,
   "require": {
     "php": "^8.0",
     "ext-json": "*",
     "ext-opentelemetry": "*",
     "laravel/framework": ">=6.0",
     "open-telemetry/api": "^1.0",
-    "open-telemetry/sem-conv": "^1.23"
+    "open-telemetry/sem-conv": "^1.24"
   },
   "require-dev": {
-    "friendsofphp/php-cs-fixer": "^3",
+    "friendsofphp/php-cs-fixer": "^3.50",
     "guzzlehttp/guzzle": "*",
-    "laravel/tinker": "*",
     "nunomaduro/collision": "*",
     "open-telemetry/sdk": "^1.0",
     "orchestra/testbench": ">=4.0",

src/Instrumentation/Laravel/src/Hooks/HookInstance.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\Contrib\Instrumentation\Laravel\Hooks;
+
+use OpenTelemetry\API\Instrumentation\CachedInstrumentation;
+
+trait HookInstance
+{
+    private static ?self $instance = null;
+
+    protected function __construct(
+        protected CachedInstrumentation $instrumentation,
+    ) {
+    }
+
+    abstract public function instrument(): void;
+
+    public static function hook(CachedInstrumentation $instrumentation): self
+    {
+        if (self::$instance === null) {
+            self::$instance = new self($instrumentation);
+            self::$instance->instrument();
+        }
+
+        return self::$instance;
+    }
+}

src/Instrumentation/Laravel/src/Hooks/Illuminate/Contracts/Queue/Queue.php

@@ -0,0 +1,145 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\Contrib\Instrumentation\Laravel\Hooks\Illuminate\Contracts\Queue;
+
+use DateInterval;
+use DateTime;
+use DateTimeInterface;
+use Illuminate\Contracts\Queue\Queue as QueueContract;
+use OpenTelemetry\API\Trace\SpanKind;
+use OpenTelemetry\Context\Context;
+use OpenTelemetry\Contrib\Instrumentation\Laravel\Hooks\HookInstance;
+use OpenTelemetry\Contrib\Instrumentation\Laravel\Hooks\Illuminate\Queue\AttributesBuilder;
+use OpenTelemetry\Contrib\Instrumentation\Laravel\Hooks\PostHookHandler;
+use function OpenTelemetry\Instrumentation\hook;
+use OpenTelemetry\SDK\Common\Time\ClockFactory;
+use OpenTelemetry\SDK\Common\Time\ClockInterface;
+use OpenTelemetry\SemConv\TraceAttributes;
+use OpenTelemetry\SemConv\TraceAttributeValues;
+use Throwable;
+
+class Queue
+{
+    use AttributesBuilder;
+    use HookInstance;
+    use PostHookHandler;
+
+    public function instrument(): void
+    {
+        $this->hookBulk();
+        $this->hookLater();
+        $this->hookPushRaw();
+    }
+
+    protected function hookBulk(): bool
+    {
+        return hook(
+            QueueContract::class,
+            'bulk',
+            pre: function (QueueContract $queue, array $params, string $class, string $function, ?string $filename, ?int $lineno) {
+                $attributes = array_merge([
+                    TraceAttributes::CODE_FUNCTION => $function,
+                    TraceAttributes::CODE_NAMESPACE => $class,
+                    TraceAttributes::CODE_FILEPATH => $filename,
+                    TraceAttributes::CODE_LINENO => $lineno,
+                    TraceAttributes::MESSAGING_BATCH_MESSAGE_COUNT => count($params[0] ?? []),
+                ], $this->contextualMessageSystemAttributes($queue, []));
+
+                /** @psalm-suppress ArgumentTypeCoercion */
+                $span = $this->instrumentation
+                    ->tracer()
+                    ->spanBuilder(vsprintf('%s %s', [
+                        method_exists($queue, 'getQueue') ? $queue->getQueue($params[2] ?? null) : $queue->getConnectionName(),
+                        TraceAttributeValues::MESSAGING_OPERATION_PUBLISH,
+                    ]))
+                    ->setSpanKind(SpanKind::KIND_PRODUCER)
+                    ->setAttributes($attributes)
+                    ->startSpan();
+
+                Context::storage()->attach($span->storeInContext(Context::getCurrent()));
+
+                return $params;
+            },
+            post: function (QueueContract $queue, array $params, $returnValue, ?Throwable $exception) {
+                $this->endSpan($exception);
+            },
+        );
+    }
+
+    protected function hookLater(): bool
+    {
+        return hook(
+            QueueContract::class,
+            'later',
+            pre: function (QueueContract $queue, array $params, string $class, string $function, ?string $filename, ?int $lineno) {
+                $clock = ClockFactory::getDefault();
+                $now = fn (): DateTimeInterface => new DateTime('@' . ($clock->now() / ClockInterface::NANOS_PER_SECOND));
+
+                $estimateDeliveryTimestamp = match (true) {
+                    is_int($params[0]) => $now()->add(new DateInterval("PT{$params[0]}S"))->getTimestamp(),
+                    $params[0] instanceof DateInterval => $now()->add($params[0])->getTimestamp(),
+                    $params[0] instanceof DateTimeInterface => ($params[0])->getTimestamp(),
+                    default => $params[0],
+                };
+
+                $attributes = [
+                    TraceAttributes::CODE_FUNCTION => $function,
+                    TraceAttributes::CODE_NAMESPACE => $class,
+                    TraceAttributes::CODE_FILEPATH => $filename,
+                    TraceAttributes::CODE_LINENO => $lineno,
+                    'messaging.message.delivery_timestamp' => $estimateDeliveryTimestamp,
+                ];
+
+                /** @psalm-suppress ArgumentTypeCoercion */
+                $span = $this->instrumentation
+                    ->tracer()
+                    ->spanBuilder(vsprintf('%s %s', [
+                        method_exists($queue, 'getQueue') ? $queue->getQueue($params[2] ?? null) : $queue->getConnectionName(),
+                        'create',
+                    ]))
+                    ->setSpanKind(SpanKind::KIND_PRODUCER)
+                    ->setAttributes($attributes)
+                    ->startSpan();
+
+                Context::storage()->attach($span->storeInContext(Context::getCurrent()));
+
+                return $params;
+            },
+            post: function (QueueContract $queue, array $params, $returnValue, ?Throwable $exception) {
+                $this->endSpan($exception);
+            },
+        );
+    }
+
+    protected function hookPushRaw(): bool
+    {
+        return hook(
+            QueueContract::class,
+            'pushRaw',
+            pre: function (QueueContract $queue, array $params, string $class, string $function, ?string $filename, ?int $lineno) {
+                $attributes = $this->buildMessageAttributes($queue, ...$params);
+
+                $parent = Context::getCurrent();
+                /** @psalm-suppress ArgumentTypeCoercion */
+                $span = $this->instrumentation
+                    ->tracer()
+                    ->spanBuilder(vsprintf('%s %s', [
+                        $attributes[TraceAttributes::MESSAGING_DESTINATION_NAME],
+                        TraceAttributeValues::MESSAGING_OPERATION_CREATE,
+                    ]))
+                    ->setSpanKind(SpanKind::KIND_PRODUCER)
+                    ->setAttributes($attributes)
+                    ->startSpan();
+
+                Context::storage()->attach($span->storeInContext($parent));
+
+                return $params;
+            },
+            post: function (QueueContract $queue, array $params, $returnValue, ?Throwable $exception) {
+                $this->endSpan($exception);
+            },
+        );
+    }
+}

src/Instrumentation/Laravel/src/Hooks/Illuminate/Foundation/Application.php

@@ -0,0 +1,44 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\Contrib\Instrumentation\Laravel\Hooks\Illuminate\Foundation;
+
+use Illuminate\Contracts\Foundation\Application as ApplicationContract;
+use Illuminate\Foundation\Application as FoundationalApplication;
+use OpenTelemetry\Contrib\Instrumentation\Laravel\Hooks\HookInstance;
+use OpenTelemetry\Contrib\Instrumentation\Laravel\Watchers\CacheWatcher;
+use OpenTelemetry\Contrib\Instrumentation\Laravel\Watchers\ClientRequestWatcher;
+use OpenTelemetry\Contrib\Instrumentation\Laravel\Watchers\ExceptionWatcher;
+use OpenTelemetry\Contrib\Instrumentation\Laravel\Watchers\LogWatcher;
+use OpenTelemetry\Contrib\Instrumentation\Laravel\Watchers\QueryWatcher;
+use OpenTelemetry\Contrib\Instrumentation\Laravel\Watchers\RequestWatcher;
+use OpenTelemetry\Contrib\Instrumentation\Laravel\Watchers\Watcher;
+use function OpenTelemetry\Instrumentation\hook;
+use Throwable;
+
+class Application
+{
+    use HookInstance;
+
+    public function instrument(): void
+    {
+        hook(
+            FoundationalApplication::class,
+            '__construct',
+            post: function (FoundationalApplication $application, array $params, mixed $returnValue, ?Throwable $exception) {
+                $this->registerWatchers($application, new CacheWatcher());
+                $this->registerWatchers($application, new ClientRequestWatcher($this->instrumentation));
+                $this->registerWatchers($application, new ExceptionWatcher());
+                $this->registerWatchers($application, new LogWatcher());
+                $this->registerWatchers($application, new QueryWatcher($this->instrumentation));
+                $this->registerWatchers($application, new RequestWatcher());
+            },
+        );
+    }
+
+    private function registerWatchers(ApplicationContract $app, Watcher $watcher): void
+    {
+        $watcher->register($app);
+    }
+}

src/Instrumentation/Laravel/src/Hooks/Illuminate/Foundation/Console/ServeCommand.php

@@ -0,0 +1,36 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\Contrib\Instrumentation\Laravel\Hooks\Illuminate\Foundation\Console;
+
+use Illuminate\Foundation\Console\ServeCommand as FoundationServeCommand;
+use OpenTelemetry\Contrib\Instrumentation\Laravel\Hooks\HookInstance;
+use function OpenTelemetry\Instrumentation\hook;
+
+/**
+ * Instrument Laravel's local PHP development server.
+ */
+class ServeCommand
+{
+    use HookInstance;
+
+    public function instrument(): void
+    {
+        hook(
+            FoundationServeCommand::class,
+            'handle',
+            pre: static function (FoundationServeCommand $serveCommand, array $params, string $class, string $function, ?string $filename, ?int $lineno) {
+                if (!property_exists(FoundationServeCommand::class, 'passthroughVariables')) {
+                    return;
+                }
+
+                foreach ($_ENV as $key => $value) {
+                    if (str_starts_with($key, 'OTEL_') && !in_array($key, FoundationServeCommand::$passthroughVariables)) {
+                        FoundationServeCommand::$passthroughVariables[] = $key;
+                    }
+                }
+            },
+        );
+    }
+}

src/Instrumentation/Laravel/src/Hooks/Illuminate/Queue/AttributesBuilder.php

@@ -0,0 +1,76 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\Contrib\Instrumentation\Laravel\Hooks\Illuminate\Queue;
+
+use Illuminate\Contracts\Queue\Queue as QueueContract;
+use Illuminate\Queue\BeanstalkdQueue;
+use Illuminate\Queue\RedisQueue;
+use Illuminate\Queue\SqsQueue;
+use OpenTelemetry\SemConv\TraceAttributes;
+use OpenTelemetry\SemConv\TraceAttributeValues;
+
+trait AttributesBuilder
+{
+    private function buildMessageAttributes(
+        QueueContract $queue,
+        string $rawPayload,
+        string $queueName = null,
+        array $options = [],
+        mixed ...$params,
+    ): array {
+        $payload = json_decode($rawPayload, true) ?? [];
+
+        return array_merge([
+            TraceAttributes::MESSAGING_DESTINATION_NAME => '(anonymous)',
+            TraceAttributes::MESSAGING_MESSAGE_ID => $payload['uuid'] ?? $payload['id'] ?? null,
+            TraceAttributes::MESSAGING_MESSAGE_ENVELOPE_SIZE => strlen($rawPayload),
+            'messaging.message.job_name' => $payload['displayName'] ?? $payload['job'] ?? null,
+            'messaging.message.attempts' => $payload['attempts'] ?? 0,
+            'messaging.message.max_exceptions' => $payload['maxExceptions'] ?? null,
+            'messaging.message.max_tries' => $payload['maxTries'] ?? null,
+            'messaging.message.retry_until' => $payload['retryUntil'] ?? null,
+            'messaging.message.timeout' => $payload['timeout'] ?? null,
+        ], $this->contextualMessageSystemAttributes($queue, $payload, $queueName, $options, ...$params));
+    }
+
+    private function contextualMessageSystemAttributes(
+        QueueContract $queue,
+        array $payload,
+        string $queueName = null,
+        array $options = [],
+        mixed ...$params,
+    ): array {
+        return match (true) {
+            $queue instanceof BeanstalkdQueue => $this->beanstalkContextualAttributes($queue, $payload, $queueName, $options, ...$params),
+            $queue instanceof RedisQueue => $this->redisContextualAttributes($queue, $payload, $queueName, $options, ...$params),
+            $queue instanceof SqsQueue => $this->awsSqsContextualAttributes($queue, $payload, $queueName, $options, ...$params),
+            default => [],
+        };
+    }
+
+    private function beanstalkContextualAttributes(BeanstalkdQueue $queue, array $payload, string $queueName = null, array $options = [], mixed ...$params): array
+    {
+        return [
+            TraceAttributes::MESSAGING_SYSTEM => 'beanstalk',
+            TraceAttributes::MESSAGING_DESTINATION_NAME => $queue->getQueue($queueName),
+        ];
+    }
+
+    private function redisContextualAttributes(RedisQueue $queue, array $payload, string $queueName = null, array $options = [], mixed ...$params): array
+    {
+        return [
+            TraceAttributes::MESSAGING_SYSTEM => 'redis',
+            TraceAttributes::MESSAGING_DESTINATION_NAME => $queue->getQueue($queueName),
+        ];
+    }
+
+    private function awsSqsContextualAttributes(SqsQueue $queue, array $payload, string $queueName = null, array $options = [], mixed ...$params): array
+    {
+        return [
+            TraceAttributes::MESSAGING_SYSTEM => TraceAttributeValues::MESSAGING_SYSTEM_AWS_SQS,
+            TraceAttributes::MESSAGING_DESTINATION_NAME => $queue->getQueue($queueName),
+        ];
+    }
+}

src/Instrumentation/Laravel/src/Hooks/Illuminate/Queue/Queue.php

@@ -0,0 +1,35 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\Contrib\Instrumentation\Laravel\Hooks\Illuminate\Queue;
+
+use Illuminate\Queue\Queue as AbstractQueue;
+use OpenTelemetry\API\Trace\Propagation\TraceContextPropagator;
+use OpenTelemetry\Contrib\Instrumentation\Laravel\Hooks\HookInstance;
+use function OpenTelemetry\Instrumentation\hook;
+use Throwable;
+
+class Queue
+{
+    use AttributesBuilder;
+    use HookInstance;
+
+    public function instrument(): void
+    {
+        $this->hookAbstractQueueCreatePayloadArray();
+    }
+
+    protected function hookAbstractQueueCreatePayloadArray(): bool
+    {
+        return hook(
+            AbstractQueue::class,
+            'createPayloadArray',
+            post: function (AbstractQueue $queue, array $params, array $payload, ?Throwable $exception): array {
+                TraceContextPropagator::getInstance()->inject($payload);
+
+                return $payload;
+            },
+        );
+    }
+}