Add integrationName to fix _dd.integration tag in push subscription

Ensures pubsub.delivery spans are properly tagged with the correct
integration name for test validation and proper categorization.

Author: nina9753

packages/datadog-plugin-google-cloud-pubsub/src/pubsub-push-subscription.js

@@ -71,11 +71,14 @@ class GoogleCloudPubsubPushSubscriptionPlugin extends TracingPlugin {
   }
 
   _createDeliverySpan (messageData, parentContext, tracer) {
-    const { message, subscription, topicName } = messageData
+    const { message, subscription, topicName, attrs } = messageData
     const subscriptionName = subscription.split('/').pop() || subscription
+    const publishStartTime = attrs['x-dd-publish-start-time']
+    const startTime = publishStartTime ? Number.parseInt(publishStartTime, 10) : undefined
 
     const span = tracer._tracer.startSpan('pubsub.delivery', {
       childOf: parentContext,
+      startTime,
       integrationName: 'google-cloud-pubsub',
       tags: {
         'span.kind': 'consumer',
@@ -93,10 +96,35 @@ class GoogleCloudPubsubPushSubscriptionPlugin extends TracingPlugin {
     })
 
     span.setTag('resource.name', `Push Subscription ${subscriptionName}`)
+    this._addBatchMetadata(span, attrs)
 
     return span
   }
 
+  _addBatchMetadata (span, attrs) {
+    const batchSize = attrs['_dd.batch.size']
+    const batchIndex = attrs['_dd.batch.index']
+
+    if (batchSize && batchIndex !== undefined) {
+      const size = Number.parseInt(batchSize, 10)
+      const index = Number.parseInt(batchIndex, 10)
+
+      span.setTag('pubsub.batch.message_count', size)
+      span.setTag('pubsub.batch.message_index', index)
+      span.setTag('pubsub.batch.description', `Message ${index + 1} of ${size}`)
+
+      const requestTraceId = attrs['_dd.pubsub_request.trace_id']
+      const requestSpanId = attrs['_dd.pubsub_request.span_id']
+
+      if (requestTraceId) {
+        span.setTag('pubsub.batch.request_trace_id', requestTraceId)
+      }
+      if (requestSpanId) {
+        span.setTag('pubsub.batch.request_span_id', requestSpanId)
+      }
+    }
+  }
+
   _extractProjectTopic (attrs, subscription) {
     const topicName = attrs['pubsub.topic']
     const projectId = subscription.match(/projects\/([^\\/]+)\/subscriptions/)