feat: otel updates for latest spec doc

.gitignore

@@ -5,7 +5,7 @@
 /.nyc_output
 /docs/
 /out/
-/build/
+**/build/
 system-test/secrets.js
 system-test/*key.json
 *.lock

src/publisher/message-queues.ts

@@ -111,6 +111,7 @@ export abstract class MessageQueue extends EventEmitter {
       }
       spanMessages.forEach(m => {
         tracing.PubsubSpans.updatePublisherTopicName(m.parentSpan!, topic.name);
+        tracing.PubsubEvents.publishStart(m);
       });
     }
 
@@ -143,6 +144,7 @@ export abstract class MessageQueue extends EventEmitter {
         // We're finished with both the RPC and the whole publish operation,
         // so close out all of the related spans.
         rpcSpan?.end();
+        tracing.PubsubEvents.publishEnd(m);
         m.parentSpan?.end();
       });
     }

src/publisher/pubsub-message.ts

@@ -55,6 +55,23 @@ export interface PubsubMessage
    * @private
    */
   publishSchedulerSpan?: tracing.Span;
+
+  /**
+   * If this is a message being received from a subscription, expose the ackId.
+   *
+   * @private
+   * @internal
+   */
+  ackId?: string;
+
+  /**
+   * If this is a message being received from a subscription, expose the exactly
+   * once delivery flag.
+   *
+   * @private
+   * @internal
+   */
+  isExactlyOnceDelivery?: boolean;
 }
 
 /**

src/subscriber.ts

@@ -135,6 +135,9 @@ export class SubscriberSpans {
   }
 
   // Emit an event for calling modAck.
+  // Note that we don't currently support users calling modAck directly, but
+  // this may be used in the future for things like fully managed pull
+  // subscriptions.
   modAckCall(deadline: Duration) {
     if (this.processing) {
       tracing.PubsubEvents.modAckCalled(this.processing, deadline);
@@ -243,6 +246,16 @@ export class Message implements tracing.MessageWithAttributes {
    */
   parentSpan?: tracing.Span;
 
+  /**
+   * We'll save the state of the subscription's exactly once delivery flag at the
+   * time the message was received. This is pretty much only for tracing, as we will
+   * generally use the live state of the subscription to figure out how to respond.
+   *
+   * @private
+   * @internal
+   */
+  isExactlyOnceDelivery: boolean;
+
   /**
    * @private
    *
@@ -344,6 +357,14 @@ export class Message implements tracing.MessageWithAttributes {
      */
     this.subSpans = new SubscriberSpans(this);
 
+    /**
+     * Save the state of the subscription into the message for later tracing.
+     *
+     * @private
+     * @internal
+     */
+    this.isExactlyOnceDelivery = sub.isExactlyOnceDelivery;
+
     this._handled = false;
     this._length = this.data.length;
     this._subscriber = sub;
@@ -423,6 +444,7 @@ export class Message implements tracing.MessageWithAttributes {
 
   /**
    * Modifies the ack deadline.
+   * At present time, this should generally not be called by users.
    *
    * @param {number} deadline The number of seconds to extend the deadline.
    * @private
@@ -437,6 +459,7 @@ export class Message implements tracing.MessageWithAttributes {
   /**
    * Modifies the ack deadline, expecting a response (for exactly-once delivery subscriptions).
    * If exactly-once delivery is not enabled, this will immediately resolve successfully.
+   * At present time, this should generally not be called by users.
    *
    * @param {number} deadline The number of seconds to extend the deadline.
    * @private

src/subscription.ts

@@ -50,7 +50,6 @@ import {DebugMessage} from './debug';
 export {AckError, AckResponse, AckResponses} from './subscriber';
 
 import {EmitterCallback, WrappingEmitter} from './wrapping-emitter';
-import * as tracing from './telemetry-tracing';
 
 export type PushConfig = google.pubsub.v1.IPushConfig;
 export type OidcToken = google.pubsub.v1.PushConfig.IOidcToken;
@@ -1238,6 +1237,7 @@ export class Subscription extends WrappingEmitter {
 
     return formatted as google.pubsub.v1.ISubscription;
   }
+
   /*!
    * Format the name of a subscription. A subscription's full name is in the
    * format of projects/{projectId}/subscriptions/{subName}.