Adds tags for TracingKafkaListener (#401)

Co-authored-by: Deeka Wong <[email protected]>
friendsofhyperf · Nov 3, 2023 · af4410d · af4410d
1 parent 5dfc3ed
commit af4410d
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 1 deletion.
diff --git a/src/sentry/publish/sentry.php b/src/sentry/publish/sentry.php
@@ -138,6 +138,9 @@
                 'exception.stack_trace' => 'exception.stack_trace',
             ],
             'kafka' => [
+                'topic' => 'kafka.topic',
+                'group_id' => 'kafka.group_id',
+                'pool' => 'kafka.pool',
                 'exception.stack_trace' => 'exception.stack_trace',
             ],
             'redis' => [

diff --git a/src/sentry/src/Tracing/Listener/TracingKafkaListener.php b/src/sentry/src/Tracing/Listener/TracingKafkaListener.php
@@ -56,12 +56,26 @@ protected function startTransaction(BeforeConsume $event): void
         $sentry = SentrySdk::init();
         $consumer = $event->getConsumer();
         $context = new TransactionContext();
-        $context->setName($consumer->getName());
+        $context->setName($consumer->getTopic());
         $context->setSource(TransactionSource::custom());
         $context->setOp('kafka.consume');
         $context->setDescription($consumer::class);
         $context->setStartTimestamp(microtime(true));
 
+        $tags = [];
+
+        if ($this->tagManager->has('kafka.topic')) {
+            $tags[$this->tagManager->get('kafka.topic')] = $consumer->getTopic();
+        }
+        if ($this->tagManager->has('kafka.group_id')) {
+            $tags[$this->tagManager->get('kafka.group_id')] = $consumer->getGroupId();
+        }
+        if ($this->tagManager->has('kafka.pool')) {
+            $tags[$this->tagManager->get('kafka.pool')] = (string) $consumer->getPool();
+        }
+
+        $context->setTags($tags);
+
         $transaction = $sentry->startTransaction($context);
 
         // If this transaction is not sampled, we can stop here to prevent doing work for nothing