Add OTEL operation attributes to queue spans

Author: onurtemizkan

dev-packages/e2e-tests/test-applications/supabase-nextjs/tests/performance.test.ts

@@ -306,6 +306,7 @@ test('Sends queue publish spans with `schema(...).rpc(...)`', async ({ page, bas
     data: {
       'messaging.destination.name': 'todos',
       'messaging.message.id': '1',
+      'messaging.message.body.size': expect.any(Number),
     },
   });
 });
@@ -355,6 +356,7 @@ test('Sends queue publish spans with `rpc(...)`', async ({ page, baseURL }) => {
     data: {
       'messaging.destination.name': 'todos',
       'messaging.message.id': '2',
+      'messaging.message.body.size': expect.any(Number),
     },
   });
 });
@@ -459,6 +461,7 @@ test('Sends queue process spans with `schema(...).rpc(...)`', async ({ page, bas
     data: {
       'messaging.destination.name': 'todos',
       'messaging.message.id': '1',
+      'messaging.message.body.size': expect.any(Number),
     },
   });
 });
@@ -563,6 +566,7 @@ test('Sends queue process spans with `rpc(...)`', async ({ page, baseURL }) => {
     data: {
       'messaging.destination.name': 'todos',
       'messaging.message.id': '2',
+      'messaging.message.body.size': expect.any(Number),
     },
   });
 });
@@ -683,6 +687,7 @@ test('Sends queue batch publish spans with `rpc(...)`', async ({ page, baseURL }
       'messaging.destination.name': 'todos',
       'messaging.message.id': expect.stringMatching(/^\d+,\d+$/),
       'messaging.batch.message_count': 2,
+      'messaging.message.body.size': expect.any(Number),
     },
   });
 });

packages/core/src/integrations/supabase.ts

@@ -491,9 +491,15 @@ function _calculateBatchLatency(messages: Array<{ enqueued_at?: string }>): numb
  * @param span - The span to process
  * @param res - The Supabase response
  * @param queueName - The name of the queue
+ * @param operationName - The queue operation name (e.g., 'pop', 'read', 'receive')
  * @returns The original response
  */
-function _processConsumerSpan(span: Span, res: SupabaseResponse, queueName: string | undefined): SupabaseResponse {
+function _processConsumerSpan(
+  span: Span,
+  res: SupabaseResponse,
+  queueName: string | undefined,
+  operationName: string,
+): SupabaseResponse {
   // Calculate latency for single message or batch average
   let latency: number | undefined;
   const isBatch = Array.isArray(res.data) && res.data.length > 1;
@@ -522,6 +528,10 @@ function _processConsumerSpan(span: Span, res: SupabaseResponse, queueName: stri
 
   // Note: messaging.destination.name is already set in initial span attributes
 
+  // Set OTEL messaging semantic attributes
+  span.setAttribute('messaging.operation.type', 'process');
+  span.setAttribute('messaging.operation.name', operationName);
+
   if (latency !== undefined) {
     span.setAttribute('messaging.message.receive.latency', latency);
   }
@@ -540,6 +550,7 @@ function _processConsumerSpan(span: Span, res: SupabaseResponse, queueName: stri
   const breadcrumbData: Record<string, unknown> = {};
   if (messageId) breadcrumbData['messaging.message.id'] = messageId;
   if (queueName) breadcrumbData['messaging.destination.name'] = queueName;
+  if (messageBodySize !== undefined) breadcrumbData['messaging.message.body.size'] = messageBodySize;
   _createQueueBreadcrumb('queue.process', queueName, breadcrumbData);
 
   // Handle errors in the response
@@ -666,7 +677,7 @@ const _instrumentRpcConsumer = (target: unknown, thisArg: unknown, argumentsList
           },
           span => {
             try {
-              const processedResponse = _processConsumerSpan(span, res, queueName);
+              const processedResponse = _processConsumerSpan(span, res, queueName, operationName);
 
               DEBUG_BUILD && debug.log('Consumer span processed successfully', { queueName });
 
@@ -788,6 +799,8 @@ function _instrumentRpcProducer(target: unknown, thisArg: unknown, argumentsList
         [SEMANTIC_ATTRIBUTE_SENTRY_OP]: 'queue.publish',
         'messaging.system': 'supabase',
         'messaging.destination.name': queueName,
+        'messaging.operation.name': operationName,
+        'messaging.operation.type': 'publish',
         ...(messageBodySize !== undefined && { 'messaging.message.body.size': messageBodySize }),
       },
     },

packages/core/test/lib/integrations/supabase-queues.test.ts

@@ -890,9 +890,49 @@ describe('Supabase Queue Instrumentation', () => {
           name: 'publish attr-test-queue',
           op: 'queue.publish',
           attributes: expect.objectContaining({
-            [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: 'auto.queue.supabase.producer',
+            [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: 'auto.db.supabase.queue.producer',
             'messaging.system': 'supabase',
             'messaging.destination.name': 'attr-test-queue',
+            'messaging.operation.name': 'send',
+            'messaging.operation.type': 'publish',
+            'messaging.message.body.size': expect.any(Number),
+          }),
+        }),
+      );
+    });
+
+    it('should set correct attributes on producer span for batch send', async () => {
+      const mockResponse: SupabaseResponse = {
+        data: [{ msg_id: 790 }, { msg_id: 791 }],
+        status: 200,
+      };
+
+      mockRpcFunction.mockResolvedValue(mockResponse);
+      instrumentSupabaseClient(mockSupabaseClient);
+
+      const startSpanSpy = vi.spyOn(Tracing, 'startSpan');
+
+      await startSpan({ name: 'test-transaction' }, async () => {
+        await mockSupabaseClient.rpc('send_batch', {
+          queue_name: 'attr-test-queue-batch',
+          messages: [{ test: 'data1' }, { test: 'data2' }],
+        });
+      });
+
+      // Find the queue.publish span call
+      const publishSpanCall = startSpanSpy.mock.calls.find(call => call[0]?.name === 'publish attr-test-queue-batch');
+
+      expect(publishSpanCall).toBeDefined();
+      expect(publishSpanCall?.[0]).toEqual(
+        expect.objectContaining({
+          name: 'publish attr-test-queue-batch',
+          op: 'queue.publish',
+          attributes: expect.objectContaining({
+            [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: 'auto.db.supabase.queue.producer',
+            'messaging.system': 'supabase',
+            'messaging.destination.name': 'attr-test-queue-batch',
+            'messaging.operation.name': 'send_batch',
+            'messaging.operation.type': 'publish',
             'messaging.message.body.size': expect.any(Number),
           }),
         }),
@@ -929,7 +969,7 @@ describe('Supabase Queue Instrumentation', () => {
           name: 'process consumer-attr-queue',
           op: 'queue.process',
           attributes: expect.objectContaining({
-            [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: 'auto.queue.supabase.consumer',
+            [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: 'auto.db.supabase.queue.consumer',
             'messaging.system': 'supabase',
             'messaging.destination.name': 'consumer-attr-queue',
           }),