feat: Opentelemetry integration

package.json

@@ -1,7 +1,7 @@
 {
   "name": "@google-cloud/pubsub",
   "description": "Cloud Pub/Sub Client Library for Node.js",
-  "version": "2.3.0",
+  "version": "2.2.0",
   "license": "Apache-2.0",
   "author": "Google Inc.",
   "engines": {
@@ -44,15 +44,15 @@
     "predocs-test": "npm run docs",
     "benchwrapper": "node bin/benchwrapper.js",
     "prelint": "cd samples; npm link ../; npm install",
-    "precompile": "gts clean",
-    "api-extractor": "api-extractor run --local",
-    "api-documenter": "api-documenter yaml --input-folder=temp"
+    "precompile": "gts clean"
   },
   "dependencies": {
     "@google-cloud/paginator": "^3.0.0",
     "@google-cloud/precise-date": "^2.0.0",
     "@google-cloud/projectify": "^2.0.0",
     "@google-cloud/promisify": "^2.0.0",
+    "@opentelemetry/api": "^0.9.0",
+    "@opentelemetry/tracing": "^0.9.0",
     "@types/duplexify": "^3.6.0",
     "@types/long": "^4.0.0",
     "arrify": "^2.0.0",
@@ -97,8 +97,6 @@
     "uuid": "^8.0.0",
     "webpack": "^4.42.0",
     "webpack-cli": "^3.3.11",
-    "yargs": "^15.0.0",
-    "@microsoft/api-documenter": "^7.8.10",
-    "@microsoft/api-extractor": "^7.8.10"
+    "yargs": "^15.0.0"
   }
 }

samples/opentelemetryTracing.js

@@ -0,0 +1,97 @@
+/*!
+ * Copyright 2020 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * This sample demonstrates how to add OpenTelemetry tracing to the
+ * Google Cloud Pub/Sub API.
+ *
+ * For more information, see the README.md under /pubsub and the documentation
+ * at https://cloud.google.com/pubsub/docs.
+ */
+
+'use strict';
+
+// sample-metadata:
+//   title: OpenTelemetry Tracing
+//   description: Demonstrates how to enable OpenTelemetry tracing in
+//     a publisher or subscriber.
+//   usage: node opentelemetryTracing.js <topic-name> <subscription-name>
+
+const SUBSCRIBER_TIMEOUT = 10;
+
+function main(
+  topicName = 'YOUR_TOPIC_NAME',
+  subscriptionName = 'YOUR_SUBSCRIPTION_NAME',
+  data = {foo: 'bar'}
+) {
+  // [START opentelemetry_tracing]
+  /**
+   * TODO(developer): Uncomment these variables before running the sample.
+   */
+  // const topicName = 'my-topic';
+  // const subscriptionName = 'my-subscription';
+  // const data = 'Hello, world!";
+
+  // Imports the Google Cloud client library
+  const {PubSub} = require('@google-cloud/pubsub');
+
+  // Imports the OpenTelemetry API
+  const {opentelemetry} = require('@opentelemetry/api');
+
+  // Imports the OpenTelemetry span handlers and exporter
+  const {
+    SimpleSpanProcessor,
+    BasicTracerProvider,
+    ConsoleSpanExporter,
+  } = require('@opentelemetry/tracing');
+
+  // Set up span processing and specify the console as the span exporter
+  const provider = new BasicTracerProvider();
+  const exporter = new ConsoleSpanExporter();
+  provider.addSpanProcessor(new SimpleSpanProcessor(exporter));
+
+  provider.register();
+  opentelemetry.trace.setGlobalTracerProvider(provider);
+
+  // Creates a client; cache this for further use
+  const pubSubClient = new PubSub();
+
+  async function publishMessage() {
+    // Publishes the message as a string, e.g. "Hello, world!" or JSON.stringify(someObject)
+    const dataBuffer = Buffer.from(data);
+
+    const messageId = await pubSubClient.topic(topicName).publish(dataBuffer);
+    console.log(`Message ${messageId} published.`);
+  }
+
+  async function subscriptionListen() {
+    // Message handler for subscriber
+    const messageHandler = message => {
+      console.log(`Message ${message.id} received.`);
+      message.ack();
+    };
+
+    // Listens for new messages from the topic
+    pubSubClient.subscription(subscriptionName).on('message', messageHandler);
+    setTimeout(() => {
+      pubSubClient.subscription(subscriptionName).removeAllListeners();
+    }, SUBSCRIBER_TIMEOUT * 1000);
+  }
+
+  publishMessage().then(subscriptionListen());
+}
+
+main(...process.argv.slice(2));

samples/package.json

@@ -14,7 +14,9 @@
     "test": "mocha system-test --timeout 600000"
   },
   "dependencies": {
-    "@google-cloud/pubsub": "^2.3.0"
+    "@google-cloud/pubsub": "^2.3.0",
+    "@opentelemetry/api": "^0.10.2",
+    "@opentelemetry/tracing": "^0.10.2"
   },
   "devDependencies": {
     "chai": "^4.2.0",

src/opentelemetry-tracing.ts

@@ -0,0 +1,43 @@
+/*!
+ * Copyright 2020 Google LLC
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import {Attributes, SpanContext, Span, trace} from '@opentelemetry/api';
+import {Tracer} from '@opentelemetry/tracing';
+
+/**
+ * Wrapper for creating OpenTelemetry Spans
+ *
+ * @class
+ */
+export class OpenTelemetryTracer {
+  /**
+   * Creates a new span with the given properties
+   *
+   * @param {string} spanName the name for the span
+   * @param {Attributes?} attributes an object containing the attributes to be set for the span
+   * @param {SpanContext?} parent the context of the parent span to link to the span
+   */
+  createSpan(
+    spanName: string,
+    attributes?: Attributes,
+    parent?: SpanContext
+  ): Span {
+    const tracerProvider: Tracer = trace.getTracer('default') as Tracer;
+    return tracerProvider.startSpan(spanName, {
+      parent: parent,
+      attributes: attributes,
+    });
+  }
+}

src/publisher/index.ts

@@ -17,13 +17,15 @@
 import {promisify, promisifyAll} from '@google-cloud/promisify';
 import * as extend from 'extend';
 import {CallOptions} from 'google-gax';
+import {Span} from '@opentelemetry/api';
 
 import {BatchPublishOptions} from './message-batch';
 import {Queue, OrderedQueue} from './message-queues';
 import {Topic} from '../topic';
 import {RequestCallback, EmptyCallback} from '../pubsub';
 import {google} from '../../protos/protos';
 import {defaultOptions} from '../default-options';
+import {OpenTelemetryTracer} from '../opentelemetry-tracing';
 
 export type PubsubMessage = google.pubsub.v1.IPubsubMessage;
 
@@ -60,11 +62,13 @@ export class Publisher {
   settings!: PublishOptions;
   queue: Queue;
   orderedQueues: Map<string, OrderedQueue>;
+  tracing: OpenTelemetryTracer;
   constructor(topic: Topic, options?: PublishOptions) {
     this.setOptions(options);
     this.topic = topic;
     this.queue = new Queue(this);
     this.orderedQueues = new Map();
+    this.tracing = new OpenTelemetryTracer();
   }
 
   flush(): Promise<void>;
@@ -136,7 +140,29 @@ export class Publisher {
    * @param {PublishCallback} [callback] Callback function.
    */
   publishMessage(message: PubsubMessage, callback: PublishCallback): void {
+    // Construct publisher span and set context as message attribute
     const {data, attributes = {}} = message;
+    const spanAttributes = {
+      data: message.data,
+    };
+    const span: Span = this.tracing.createSpan(
+      `${this.topic.name} publisher`,
+      spanAttributes
+    );
+    if (
+      message.attributes &&
+      message.attributes['googclient_OpenTelemetrySpanContext']
+    ) {
+      console.warn(
+        'googclient_OpenTelemetrySpanContext key set as message attribute, but will be overridden.'
+      );
+    }
+    if (!message.attributes) {
+      message.attributes = {};
+    }
+    message.attributes['googclient_OpenTelemetrySpanContext'] = JSON.stringify(
+      span.context()
+    );
 
     if (!(data instanceof Buffer)) {
       throw new TypeError('Data must be in the form of a Buffer.');
@@ -152,6 +178,7 @@ export class Publisher {
 
     if (!message.orderingKey) {
       this.queue.add(message, callback);
+      span.end();
       return;
     }
 
@@ -165,6 +192,8 @@ export class Publisher {
 
     const queue = this.orderedQueues.get(key)!;
     queue.add(message, callback);
+
+    span.end();
   }
   /**
    * Indicates to the publisher that it is safe to continue publishing for the

src/subscriber.ts

@@ -18,6 +18,7 @@ import {DateStruct, PreciseDate} from '@google-cloud/precise-date';
 import {replaceProjectIdToken} from '@google-cloud/projectify';
 import {promisify} from '@google-cloud/promisify';
 import {EventEmitter} from 'events';
+import {SpanContext} from '@opentelemetry/api';
 
 import {google} from '../protos/protos';
 import {Histogram} from './histogram';
@@ -27,6 +28,7 @@ import {MessageStream, MessageStreamOptions} from './message-stream';
 import {Subscription} from './subscription';
 import {defaultOptions} from './default-options';
 import {SubscriberClient} from './v1';
+import {OpenTelemetryTracer} from './opentelemetry-tracing';
 
 export type PullResponse = google.pubsub.v1.IPullResponse;
 
@@ -237,6 +239,7 @@ export class Subscriber extends EventEmitter {
   private _options!: SubscriberOptions;
   private _stream!: MessageStream;
   private _subscription: Subscription;
+  private _tracing: OpenTelemetryTracer;
   constructor(subscription: Subscription, options = {}) {
     super();
 
@@ -248,6 +251,7 @@ export class Subscriber extends EventEmitter {
     this._histogram = new Histogram({min: 10, max: 600});
     this._latencies = new Histogram();
     this._subscription = subscription;
+    this._tracing = new OpenTelemetryTracer();
 
     this.setOptions(options);
   }
@@ -445,12 +449,35 @@ export class Subscriber extends EventEmitter {
     for (const data of receivedMessages!) {
       const message = new Message(this, data);
 
+      // Create a new OpenTelemetry span with the publisher span set as the parent
+      const spanValue = message.attributes
+        ? message.attributes['googclient_OpenTelemetrySpanContext']
+        : null;
+
+      const parentSpanContext: SpanContext | null = spanValue
+        ? JSON.parse(spanValue)
+        : null;
+
+      const spanAttributes = {
+        ackId: data.ackId,
+        deliveryAttempt: data.deliveryAttempt,
+      };
+      const span = parentSpanContext
+        ? this._tracing.createSpan(
+            this._name,
+            spanAttributes,
+            parentSpanContext
+          )
+        : null;
       if (this.isOpen) {
         message.modAck(this.ackDeadline);
         this._inventory.add(message);
       } else {
         message.nack();
       }
+      if (span) {
+        span.end();
+      }
     }
   }
 

src/v1/publisher_client.ts

@@ -390,9 +390,8 @@ export class PublisherClient {
     >
   ): void;
   /**
-   * Creates the given topic with the given name. See the
-   * <a href="https://cloud.google.com/pubsub/docs/admin#resource_names">
-   * resource name rules</a>.
+   * Creates the given topic with the given name. See the [resource name rules](
+   * https://cloud.google.com/pubsub/docs/admin#resource_names).
    *
    * @param {Object} request
    *   The request object that will be sent.
@@ -404,8 +403,8 @@ export class PublisherClient {
    *   signs (`%`). It must be between 3 and 255 characters in length, and it
    *   must not start with `"goog"`.
    * @param {number[]} request.labels
-   *   See <a href="https://cloud.google.com/pubsub/docs/labels"> Creating and
-   *   managing labels</a>.
+   *   See [Creating and managing labels]
+   *   (https://cloud.google.com/pubsub/docs/labels).
    * @param {google.pubsub.v1.MessageStoragePolicy} request.messageStoragePolicy
    *   Policy constraining the set of Google Cloud Platform regions where messages
    *   published to the topic may be stored. If not present, then no constraints
@@ -1312,11 +1311,10 @@ export class PublisherClient {
   ): void;
   /**
    * Lists the names of the snapshots on this topic. Snapshots are used in
-   * <a href="https://cloud.google.com/pubsub/docs/replay-overview">Seek</a>
-   * operations, which allow
-   * you to manage message acknowledgments in bulk. That is, you can set the
-   * acknowledgment state of messages in an existing subscription to the state
-   * captured by a snapshot.
+   * [Seek](https://cloud.google.com/pubsub/docs/replay-overview) operations,
+   * which allow you to manage message acknowledgments in bulk. That is, you can
+   * set the acknowledgment state of messages in an existing subscription to the
+   * state captured by a snapshot.
    *
    * @param {Object} request
    *   The request object that will be sent.