Feat/sentry coroutine context propagation improvements

composer.json

@@ -32,7 +32,7 @@
         "psr/http-factory-implementation": "*",
         "psy/psysh": "^0.10.0 || ^0.11.0",
         "ramsey/uuid": "^4.7",
-        "sentry/sentry": "^4.15.0",
+        "sentry/sentry": "^4.16.0",
         "symfony/console": "^5.3 || ^6.0 || ^7.0",
         "symfony/http-foundation": "^5.3 || ^6.0 || ^7.0",
         "symfony/polyfill-php84": "^1.33",

src/sentry/composer.json

@@ -33,7 +33,7 @@
         "hyperf/http-server": "~3.1.0",
         "hyperf/support": "~3.1.0",
         "hyperf/tappable": "~3.1.0",
-        "sentry/sentry": "^4.15.0",
+        "sentry/sentry": "^4.16.0",
         "symfony/polyfill-php85": "^1.33"
     },
     "suggest": {

src/sentry/src/Aspect/CoroutineAspect.php

@@ -29,6 +29,11 @@ class CoroutineAspect extends AbstractAspect
         \Psr\Http\Message\ServerRequestInterface::class,
     ];
 
+    public function __construct()
+    {
+        $this->priority = PHP_INT_MAX - 1;
+    }
+
     public function process(ProceedingJoinPoint $proceedingJoinPoint)
     {
         $callable = $proceedingJoinPoint->arguments['keys']['callable'];
@@ -40,7 +45,7 @@ public function process(ProceedingJoinPoint $proceedingJoinPoint)
             $current = Co::getContextFor();
 
             foreach ($keys as $key) {
-                if (isset($from[$key])) {
+                if (isset($from[$key]) && ! isset($current[$key])) {
                     $current[$key] = $from[$key];
                 }
             }

src/sentry/src/Tracing/Aspect/AmqpProducerAspect.php

@@ -17,12 +17,13 @@
 use FriendsOfHyperf\Sentry\Util\Carrier;
 use Hyperf\Amqp\Annotation\Producer;
 use Hyperf\Amqp\Message\ProducerMessage;
+use Hyperf\Coroutine\Coroutine;
 use Hyperf\Di\Annotation\AnnotationCollector;
 use Hyperf\Di\Aop\AbstractAspect;
 use Hyperf\Di\Aop\ProceedingJoinPoint;
 use PhpAmqpLib\Wire\AMQPTable;
-
-use function Hyperf\Tappable\tap;
+use Sentry\State\Scope;
+use Sentry\Tracing\SpanContext;
 
 /**
  * @property array{application_headers?:AMQPTable} $properties
@@ -60,16 +61,6 @@ protected function handleProduceMessage(ProceedingJoinPoint $proceedingJoinPoint
             return $proceedingJoinPoint->process();
         }
 
-        $span = $this->startSpan(
-            op: 'queue.publish',
-            description: sprintf('%s::%s()', $proceedingJoinPoint->className, $proceedingJoinPoint->methodName),
-            origin: 'auto.amqp'
-        );
-
-        if (! $span) {
-            return $proceedingJoinPoint->process();
-        }
-
         $routingKey = $producerMessage->getRoutingKey();
         $exchange = $producerMessage->getExchange();
         $poolName = $producerMessage->getPoolName();
@@ -87,30 +78,42 @@ protected function handleProduceMessage(ProceedingJoinPoint $proceedingJoinPoint
         $messageId = uniqid('amqp_', true);
         $destinationName = implode(', ', (array) $routingKey);
         $bodySize = strlen($producerMessage->payload());
-        $span->setData([
-            'messaging.system' => 'amqp',
-            'messaging.operation' => 'publish',
-            'messaging.message.id' => $messageId,
-            'messaging.message.body.size' => $bodySize,
-            'messaging.destination.name' => $destinationName,
-            // for amqp
-            'messaging.amqp.message.type' => $producerMessage->getTypeString(),
-            'messaging.amqp.message.routing_key' => $routingKey,
-            'messaging.amqp.message.exchange' => $exchange,
-            'messaging.amqp.message.pool_name' => $poolName,
-        ]);
-        $carrier = Carrier::fromSpan($span)->with([
-            'publish_time' => microtime(true),
-            'message_id' => $messageId,
-            'destination_name' => $destinationName,
-            'body_size' => $bodySize,
-        ]);
 
-        (function () use ($carrier) {
-            $this->properties['application_headers'] ??= new AMQPTable();
-            $this->properties['application_headers']->set(Constants::TRACE_CARRIER, $carrier->toJson());
-        })->call($producerMessage);
-
-        return tap($proceedingJoinPoint->process(), fn () => $span->finish());
+        return $this->trace(
+            function (Scope $scope) use ($proceedingJoinPoint, $producerMessage, $messageId, $destinationName, $bodySize) {
+                $span = $scope->getSpan();
+                if ($span) {
+                    $carrier = Carrier::fromSpan($span)->with([
+                        'publish_time' => microtime(true),
+                        'message_id' => $messageId,
+                        'destination_name' => $destinationName,
+                        'body_size' => $bodySize,
+                    ]);
+                    (function () use ($carrier) {
+                        $this->properties['application_headers'] ??= new AMQPTable();
+                        $this->properties['application_headers']->set(Constants::TRACE_CARRIER, $carrier->toJson());
+                    })->call($producerMessage);
+                }
+
+                return $proceedingJoinPoint->process();
+            },
+            SpanContext::make()
+                ->setOp('queue.publish')
+                ->setDescription(sprintf('%s::%s()', $proceedingJoinPoint->className, $proceedingJoinPoint->methodName))
+                ->setOrigin('auto.amqp')
+                ->setData([
+                    'coroutine.id' => Coroutine::id(),
+                    'messaging.system' => 'amqp',
+                    'messaging.operation' => 'publish',
+                    'messaging.message.id' => $messageId,
+                    'messaging.message.body.size' => $bodySize,
+                    'messaging.destination.name' => $destinationName,
+                    // for amqp
+                    'messaging.amqp.message.type' => $producerMessage->getTypeString(),
+                    'messaging.amqp.message.routing_key' => $routingKey,
+                    'messaging.amqp.message.exchange' => $exchange,
+                    'messaging.amqp.message.pool_name' => $poolName,
+                ])
+        );
     }
 }

src/sentry/src/Tracing/Aspect/AsyncQueueJobMessageAspect.php

@@ -17,8 +17,11 @@
 use FriendsOfHyperf\Sentry\Util\Carrier;
 use Hyperf\AsyncQueue\Driver\RedisDriver;
 use Hyperf\Context\Context;
+use Hyperf\Coroutine\Coroutine;
 use Hyperf\Di\Aop\AbstractAspect;
 use Hyperf\Di\Aop\ProceedingJoinPoint;
+use Sentry\State\Scope;
+use Sentry\Tracing\SpanContext;
 
 use function Hyperf\Support\with;
 
@@ -73,46 +76,45 @@ public function handlePush(ProceedingJoinPoint $proceedingJoinPoint)
     {
         /** @var \Hyperf\AsyncQueue\JobInterface $job */
         $job = $proceedingJoinPoint->arguments['keys']['job'] ?? null;
-        $span = $this->startSpan(
-            op: 'queue.publish',
-            description: $job::class,
-            origin: 'auto.queue'
-        );
 
-        if (! $span) {
-            return $proceedingJoinPoint->process();
+        /** @var \Hyperf\AsyncQueue\Driver\Driver $driver */
+        $driver = $proceedingJoinPoint->getInstance();
+        $messageId = method_exists($job, 'getId') ? $job->getId() : uniqid('async_queue_', true);
+        $destinationName = Context::get('sentry.messaging.destination.name', 'default');
+        $bodySize = (fn ($job) => strlen($this->packer->pack($job)))->call($driver, $job);
+        $data = [
+            'coroutine.id' => Coroutine::id(),
+            'messaging.system' => 'async_queue',
+            'messaging.operation' => 'publish',
+            'messaging.message.id' => $messageId,
+            'messaging.message.body.size' => $bodySize,
+            'messaging.destination.name' => $destinationName,
+        ];
+
+        if ($driver instanceof RedisDriver) {
+            $data = array_merge($data, $this->buildSpanDataOfRedisDriver($driver));
         }
 
-        try {
-            /** @var \Hyperf\AsyncQueue\Driver\Driver $driver */
-            $driver = $proceedingJoinPoint->getInstance();
-            $messageId = method_exists($job, 'getId') ? $job->getId() : uniqid('async_queue_', true);
-            $destinationName = Context::get('sentry.messaging.destination.name', 'default');
-            $bodySize = (fn ($job) => strlen($this->packer->pack($job)))->call($driver, $job);
-            $span->setData([
-                'messaging.system' => 'async_queue',
-                'messaging.operation' => 'publish',
-                'messaging.message.id' => $messageId,
-                'messaging.message.body.size' => $bodySize,
-                'messaging.destination.name' => $destinationName,
-            ]);
-            if ($driver instanceof RedisDriver) {
-                $span->setData($this->buildSpanDataOfRedisDriver($driver));
-            }
-
-            $carrier = Carrier::fromSpan($span)->with([
-                'publish_time' => microtime(true),
-                'message_id' => $messageId,
-                'destination_name' => $destinationName,
-                'body_size' => $bodySize,
-            ]);
-
-            Context::set(Constants::TRACE_CARRIER, $carrier);
-
-            return $proceedingJoinPoint->process();
-        } finally {
-            $span->finish();
-        }
+        return $this->trace(
+            function (Scope $scope) use ($proceedingJoinPoint, $messageId, $destinationName, $bodySize) {
+                $span = $scope->getSpan();
+                $carrier = Carrier::fromSpan($span)->with([
+                    'publish_time' => microtime(true),
+                    'message_id' => $messageId,
+                    'destination_name' => $destinationName,
+                    'body_size' => $bodySize,
+                ]);
+
+                Context::set(Constants::TRACE_CARRIER, $carrier);
+
+                return $proceedingJoinPoint->process();
+            },
+            SpanContext::make()
+                ->setOp('queue.publish')
+                ->setDescription($job::class)
+                ->setOrigin('auto.queue')
+                ->setData($data)
+        );
     }
 
     protected function buildSpanDataOfRedisDriver(RedisDriver $driver): array

src/sentry/src/Tracing/Aspect/CacheAspect.php

@@ -13,9 +13,11 @@
 
 use FriendsOfHyperf\Sentry\Switcher;
 use FriendsOfHyperf\Sentry\Tracing\SpanStarter;
+use Hyperf\Coroutine\Coroutine;
 use Hyperf\Di\Aop\AbstractAspect;
 use Hyperf\Di\Aop\ProceedingJoinPoint;
-use Sentry\SentrySdk;
+use Sentry\State\Scope;
+use Sentry\Tracing\SpanContext;
 
 use function Hyperf\Tappable\tap;
 
@@ -47,63 +49,60 @@ public function process(ProceedingJoinPoint $proceedingJoinPoint)
             return $proceedingJoinPoint->process();
         }
 
-        $parent = SentrySdk::getCurrentHub()->getSpan();
+        $method = $proceedingJoinPoint->methodName;
+        $op = match ($method) {
+            'set', 'setMultiple' => 'cache.put',
+            'get', 'fetch', 'getMultiple' => 'cache.get',
+            'delete', 'deleteMultiple' => 'cache.remove',
+            'clear' => 'cache.flush',
+            default => 'cache',
+        };
 
-        try {
-            $method = $proceedingJoinPoint->methodName;
-            $op = match ($method) {
-                'set', 'setMultiple' => 'cache.put',
-                'get', 'fetch', 'getMultiple' => 'cache.get',
-                'delete', 'deleteMultiple' => 'cache.remove',
-                'clear' => 'cache.flush',
-                default => 'cache',
-            };
+        $arguments = $proceedingJoinPoint->arguments['keys'] ?? [];
 
-            $arguments = $proceedingJoinPoint->arguments['keys'] ?? [];
+        /** @var string[] $keys */
+        $keys = match ($method) {
+            'set', 'get', 'delete' => [$arguments['key'] ?? 'unknown'],
+            'setMultiple' => array_keys($arguments['values'] ?? []),
+            'getMultiple', 'deleteMultiple' => $arguments['keys'] ?? [],
+            default => [],
+        };
 
-            /** @var string[] $keys */
-            $keys = match ($method) {
-                'set', 'get', 'delete' => [$arguments['key'] ?? 'unknown'],
-                'setMultiple' => array_keys($arguments['values'] ?? []),
-                'getMultiple', 'deleteMultiple' => $arguments['keys'] ?? [],
-                default => [],
-            };
-
-            $span = $this->startSpan(
-                op: $op,
-                description: implode(', ', $keys),
-                origin: 'auto.cache',
-                asParent: true
-            )?->setData([
-                'cache.key' => $keys,
-                'cache.ttl' => $arguments['ttl'] ?? null,
-                'item_size' => match (true) {
-                    isset($arguments['value']) => strlen(json_encode($arguments['value'])),
-                    isset($arguments['values']) && is_array($arguments['values']) => strlen(json_encode(array_values($arguments['values']))),
-                    default => 0,
-                },
-            ]);
-
-            return tap($proceedingJoinPoint->process(), function ($result) use ($span, $method) {
-                $data = match ($method) {
-                    'get' => [
-                        'cache.hit' => ! is_null($result),
-                        'cache.item_size' => strlen((string) json_encode($result)),
-                    ],
-                    'fetch' => [
-                        'cache.hit' => ($result[0] ?? false) !== false,
-                        'cache.item_size' => strlen((string) json_encode($result[1] ?? '')),
-                    ],
-                    'getMultiple' => [
-                        'cache.hit' => ! empty($result),
-                        'cache.item_size' => strlen((string) json_encode(array_values((array) $result))),
-                    ],
-                    default => [],
-                };
-                $span?->setData($data)->finish();
-            });
-        } finally {
-            SentrySdk::getCurrentHub()->setSpan($parent);
-        }
+        return $this->trace(
+            function (Scope $scope) use ($proceedingJoinPoint, $method) {
+                return tap($proceedingJoinPoint->process(), function ($result) use ($method, $scope) {
+                    $data = match ($method) {
+                        'get' => [
+                            'cache.hit' => ! is_null($result),
+                            'cache.item_size' => strlen((string) json_encode($result)),
+                        ],
+                        'fetch' => [
+                            'cache.hit' => ($result[0] ?? false) !== false,
+                            'cache.item_size' => strlen((string) json_encode($result[1] ?? '')),
+                        ],
+                        'getMultiple' => [
+                            'cache.hit' => ! empty($result),
+                            'cache.item_size' => strlen((string) json_encode(array_values((array) $result))),
+                        ],
+                        default => [],
+                    };
+                    $scope->getSpan()?->setData($data);
+                });
+            },
+            SpanContext::make()
+                ->setOp($op)
+                ->setDescription(implode(', ', $keys))
+                ->setOrigin('auto.cache')
+                ->setData([
+                    'coroutine.id' => Coroutine::id(),
+                    'cache.key' => $keys,
+                    'cache.ttl' => $arguments['ttl'] ?? null,
+                    'item_size' => match (true) {
+                        isset($arguments['value']) => strlen(json_encode($arguments['value'])),
+                        isset($arguments['values']) && is_array($arguments['values']) => strlen(json_encode(array_values($arguments['values']))),
+                        default => 0,
+                    },
+                ])
+        );
     }
 }