fix: set messaging.operation in createAttributes

src/telemetry-tracing.ts

@@ -305,6 +305,7 @@ export class PubsubSpans {
     params: AttributeParams,
     message?: PubsubMessage,
     caller?: string,
+    operation?: string,
   ): SpanAttributes {
     const destinationName = params.topicName ?? params.subName;
     const destinationId = params.topicId ?? params.subId;
@@ -351,6 +352,9 @@ export class PubsubSpans {
       if (message.ackId) {
         spanAttributes['messaging.gcp_pubsub.message.ack_id'] = message.ackId;
       }
+      if (operation) {
+        spanAttributes['messaging.operation'] = operation;
+      }
     }
 
     return spanAttributes;
@@ -368,11 +372,15 @@ export class PubsubSpans {
     const topicInfo = getTopicInfo(topicName);
     const span: Span = getTracer().startSpan(`${topicName} create`, {
       kind: SpanKind.PRODUCER,
-      attributes: PubsubSpans.createAttributes(topicInfo, message, caller),
+      attributes: PubsubSpans.createAttributes(
+        topicInfo,
+        message,
+        caller,
+        'create',
+      ),
     });
     if (topicInfo.topicId) {
       span.updateName(`${topicInfo.topicId} create`);
-      span.setAttribute('messaging.operation', 'create');
       span.setAttribute('messaging.destination.name', topicInfo.topicId);
     }
 
@@ -404,10 +412,14 @@ export class PubsubSpans {
 
     const subInfo = getSubscriptionInfo(subName);
     const name = `${subInfo.subId ?? subName} subscribe`;
-    const attributes = this.createAttributes(subInfo, message, caller);
+    const attributes = this.createAttributes(
+      subInfo,
+      message,
+      caller,
+      'receive',
+    );
     if (subInfo.subId) {
       attributes['messaging.destination.name'] = subInfo.subId;
-      attributes['messaging.operation'] = 'receive';
     }
 
     if (context) {
@@ -473,6 +485,7 @@ export class PubsubSpans {
       getTopicInfo(topicName),
       undefined,
       caller,
+      'create',
     );
     const links: Link[] = messages
       .filter(m => m.parentSpan && isSampled(m.parentSpan))
@@ -515,6 +528,7 @@ export class PubsubSpans {
       subInfo,
       undefined,
       caller,
+      'receive',
     );
     const links: Link[] = messageSpans
       .filter(m => m && isSampled(m))
@@ -531,7 +545,6 @@ export class PubsubSpans {
     );
 
     span?.setAttribute('messaging.batch.message_count', messageSpans.length);
-    span?.setAttribute('messaging.operation', 'receive');
 
     if (span) {
       // Also attempt to link from the subscribe span(s) back to the publish RPC span.
@@ -563,6 +576,7 @@ export class PubsubSpans {
       subInfo,
       undefined,
       caller,
+      'receive',
     );
     const links: Link[] = messageSpans
       .filter(m => m && isSampled(m))
@@ -579,7 +593,6 @@ export class PubsubSpans {
     );
 
     span?.setAttribute('messaging.batch.message_count', messageSpans.length);
-    span?.setAttribute('messaging.operation', 'receive');
 
     if (span) {
       // Also attempt to link from the subscribe span(s) back to the publish RPC span.

test/telemetry-tracing.ts

@@ -212,6 +212,7 @@ describe('OpenTelemetryTracer', () => {
         topicInfo,
         message,
         'tests',
+        'create',
       );
       assert.deepStrictEqual(topicAttrs, {
         'messaging.system': 'gcp_pubsub',
@@ -222,6 +223,7 @@ describe('OpenTelemetryTracer', () => {
         'messaging.gcp_pubsub.message.exactly_once_delivery':
           message.isExactlyOnceDelivery,
         'messaging.gcp_pubsub.message.ack_id': message.ackId,
+        'messaging.operation': 'create',
         'code.function': 'tests',
       });
 
@@ -235,10 +237,12 @@ describe('OpenTelemetryTracer', () => {
         topicInfo,
         message,
         'tests',
+        'create',
       );
       assert.deepStrictEqual(topicAttrs2, {
         'messaging.system': 'gcp_pubsub',
         'messaging.destination.name': topicInfo.topicId,
+        'messaging.operation': 'create',
         'gcp.project_id': topicInfo.projectId,
         'messaging.message.envelope.size': message.data?.length,
         'code.function': 'tests',
@@ -283,6 +287,7 @@ describe('OpenTelemetryTracer', () => {
       const firstSpan = spans.pop();
       assert.ok(firstSpan);
       assert.strictEqual(firstSpan.name, `${tests.topicInfo.topicId} create`);
+      assert.strictEqual(firstSpan.attributes['messaging.operation'], 'create');
       assert.strictEqual(
         firstSpan.attributes['messaging.destination.name'],
         tests.topicInfo.topicId,
@@ -312,7 +317,6 @@ describe('OpenTelemetryTracer', () => {
       const firstSpan = spans.pop();
       assert.ok(firstSpan);
       assert.strictEqual(firstSpan.name, 'other create');
-
       assert.strictEqual(
         firstSpan.attributes['messaging.destination.name'],
         'other',
@@ -346,6 +350,10 @@ describe('OpenTelemetryTracer', () => {
         childReadSpan.attributes['messaging.destination.name'],
         'sub',
       );
+      assert.strictEqual(
+        childReadSpan.attributes['messaging.operation'],
+        'receive',
+      );
       assert.strictEqual(childReadSpan.kind, SpanKind.CONSUMER);
       assert.ok(childReadSpan.parentSpanId);
     });