feat: Opentelemetry integration (#1078)

* Add opentelemetry tracing

* build: rename _toc to toc (#1066)

* changes without context

        autosynth cannot find the source of changes triggered by earlier changes in this
        repository, or by version upgrades to tools such as linters.

* fix: rename _toc to toc

Source-Author: F. Hinkelmann <[email protected]>
Source-Date: Tue Jul 21 10:53:20 2020 -0400
Source-Repo: googleapis/synthtool
Source-Sha: 99c93fe09f8c1dca09dfc0301c8668e3a70dd796
Source-Link: googleapis/synthtool@99c93fe

Co-authored-by: sofisl <[email protected]>

* build: move gitattributes files to node templates (#1070)

Source-Author: F. Hinkelmann <[email protected]>
Source-Date: Thu Jul 23 01:45:04 2020 -0400
Source-Repo: googleapis/synthtool
Source-Sha: 3a00b7fea8c4c83eaff8eb207f530a2e3e8e1de3
Source-Link: googleapis/synthtool@3a00b7f

* Add opentelemetry instrumentation

* Add create span test

* Refactor tracing

* Add publisher key test

* Fix linting issues

* Add docs

* Add example for opentelemetry

* Add tracing example

* Update headers

* Add microsoft api documenter

* Fix linting in samples/package.json

* Add optional tracing

* Fix linting issues

* Re-add api-documenter

* Update package.json

* Update package.json

* Update package.json

* Fix docs

* Add more unit tests

* Fix linting

* Add disable tracing tests

* Update opentelemetryTracing sample

Co-authored-by: Yoshi Automation Bot <[email protected]>
Co-authored-by: sofisl <[email protected]>
Co-authored-by: Benjamin E. Coe <[email protected]>
Co-authored-by: Megan Potter <[email protected]>

Author: 5 people

handwritten/pubsub/package.json

@@ -30,7 +30,7 @@
     "presystem-test": "npm run compile",
     "system-test": "mocha build/system-test --timeout 600000",
     "samples-test": "cd samples/ && npm link ../ && npm install && npm test && cd ../",
-    "test": "c8 mocha build/test",
+    "test": "c8 mocha build/test --recursive",
     "lint": "gts check",
     "predocs": "npm run compile",
     "docs": "jsdoc -c .jsdoc.js",
@@ -44,15 +44,17 @@
     "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"
+    "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",

handwritten/pubsub/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,
+    });
+  }
+}

handwritten/pubsub/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;
 
@@ -37,6 +39,7 @@ export interface PublishOptions {
   batching?: BatchPublishOptions;
   gaxOpts?: CallOptions;
   messageOrdering?: boolean;
+  enableOpenTelemetryTracing?: boolean;
 }
 
 /**
@@ -72,11 +75,16 @@ export class Publisher {
   settings!: PublishOptions;
   queue: Queue;
   orderedQueues: Map<string, OrderedQueue>;
+  tracing: OpenTelemetryTracer | undefined;
   constructor(topic: Topic, options?: PublishOptions) {
     this.setOptions(options);
     this.topic = topic;
     this.queue = new Queue(this);
     this.orderedQueues = new Map();
+    this.tracing =
+      this.settings && this.settings.enableOpenTelemetryTracing
+        ? new OpenTelemetryTracer()
+        : undefined;
   }
 
   flush(): Promise<void>;
@@ -162,8 +170,13 @@ export class Publisher {
       }
     }
 
+    const span: Span | undefined = this.constructSpan(message);
+
     if (!message.orderingKey) {
       this.queue.add(message, callback);
+      if (span) {
+        span.end();
+      }
       return;
     }
 
@@ -177,6 +190,10 @@ export class Publisher {
 
     const queue = this.orderedQueues.get(key)!;
     queue.add(message, callback);
+
+    if (span) {
+      span.end();
+    }
   }
   /**
    * Indicates to the publisher that it is safe to continue publishing for the
@@ -211,13 +228,19 @@ export class Publisher {
       gaxOpts: {
         isBundling: false,
       },
+      enableOpenTelemetryTracing: false,
     };
 
-    const {batching, gaxOpts, messageOrdering} = extend(
-      true,
-      defaults,
-      options
-    );
+    const {
+      batching,
+      gaxOpts,
+      messageOrdering,
+      enableOpenTelemetryTracing,
+    } = extend(true, defaults, options);
+
+    this.tracing = enableOpenTelemetryTracing
+      ? new OpenTelemetryTracer()
+      : undefined;
 
     this.settings = {
       batching: {
@@ -227,11 +250,45 @@ export class Publisher {
       },
       gaxOpts,
       messageOrdering,
+      enableOpenTelemetryTracing,
     };
   }
+
+  /**
+   * Constructs an OpenTelemetry span
+   *
+   * @private
+   *
+   * @param {PubsubMessage} message The message to create a span for
+   */
+  constructSpan(message: PubsubMessage): Span | undefined {
+    const spanAttributes = {
+      data: message.data,
+    };
+    const span: Span | undefined = this.tracing
+      ? this.tracing.createSpan(`${this.topic.name} publisher`, spanAttributes)
+      : undefined;
+    if (span) {
+      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());
+    }
+    return span;
+  }
 }
 
 promisifyAll(Publisher, {
   singular: true,
-  exclude: ['publish', 'setOptions'],
+  exclude: ['publish', 'setOptions', 'constructSpan'],
 });

handwritten/pubsub/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, Span} 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;
 
@@ -202,6 +204,7 @@ export interface SubscriberOptions {
   batching?: BatchOptions;
   flowControl?: FlowControlOptions;
   streamingOptions?: MessageStreamOptions;
+  enableOpenTelemetryTracing?: boolean;
 }
 
 /**
@@ -237,6 +240,7 @@ export class Subscriber extends EventEmitter {
   private _options!: SubscriberOptions;
   private _stream!: MessageStream;
   private _subscription: Subscription;
+  private _tracing: OpenTelemetryTracer | undefined;
   constructor(subscription: Subscription, options = {}) {
     super();
 
@@ -248,7 +252,6 @@ export class Subscriber extends EventEmitter {
     this._histogram = new Histogram({min: 10, max: 600});
     this._latencies = new Histogram();
     this._subscription = subscription;
-
     this.setOptions(options);
   }
   /**
@@ -423,7 +426,42 @@ export class Subscriber extends EventEmitter {
         this.maxMessages
       );
     }
+    this._tracing = options.enableOpenTelemetryTracing
+      ? new OpenTelemetryTracer()
+      : undefined;
+  }
+
+  /**
+   * Constructs an OpenTelemetry span from the incoming message.
+   *
+   * @param {Message} message One of the received messages
+   * @private
+   */
+  private _constructSpan(message: Message): Span | undefined {
+    // Handle cases where OpenTelemetry is disabled or no span context was sent through message
+    if (
+      !this._tracing ||
+      !message.attributes ||
+      !message.attributes['googclient_OpenTelemetrySpanContext']
+    ) {
+      return undefined;
+    }
+    const spanValue = message.attributes['googclient_OpenTelemetrySpanContext'];
+    const parentSpanContext: SpanContext | undefined = spanValue
+      ? JSON.parse(spanValue)
+      : undefined;
+    const spanAttributes = {
+      ackId: message.ackId,
+      deliveryAttempt: message.deliveryAttempt,
+    };
+    // Subscriber spans should always have a publisher span as a parent.
+    // Return undefined if no parent is provided
+    const span = parentSpanContext
+      ? this._tracing.createSpan(this._name, spanAttributes, parentSpanContext)
+      : undefined;
+    return span;
   }
+
   /**
    * Callback to be invoked when a new message is available.
    *
@@ -445,12 +483,16 @@ export class Subscriber extends EventEmitter {
     for (const data of receivedMessages!) {
       const message = new Message(this, data);
 
+      const span: Span | undefined = this._constructSpan(message);
       if (this.isOpen) {
         message.modAck(this.ackDeadline);
         this._inventory.add(message);
       } else {
         message.nack();
       }
+      if (span) {
+        span.end();
+      }
     }
   }
 

handwritten/pubsub/test/opentelemetry-tracing.ts

@@ -0,0 +1,63 @@
+/*!
+ * 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 * as assert from 'assert';
+import {describe, it, before, beforeEach, afterEach} from 'mocha';
+
+import * as api from '@opentelemetry/api';
+import * as trace from '@opentelemetry/tracing';
+import {OpenTelemetryTracer} from '../src/opentelemetry-tracing';
+import {SimpleSpanProcessor} from '@opentelemetry/tracing';
+
+describe('OpenTelemetryTracer', () => {
+  let tracing: OpenTelemetryTracer;
+  let span: trace.Span;
+  const spanName = 'test-span';
+  const spanContext: api.SpanContext = {
+    traceId: 'd4cda95b652f4a1592b449d5929fda1b',
+    spanId: '6e0c63257de34c92',
+    traceFlags: api.TraceFlags.SAMPLED,
+  };
+  const spanAttributes: api.Attributes = {
+    foo: 'bar',
+  };
+
+  before(() => {
+    const provider = new trace.BasicTracerProvider();
+    const exporter = new trace.InMemorySpanExporter();
+    provider.addSpanProcessor(new SimpleSpanProcessor(exporter));
+    api.trace.setGlobalTracerProvider(provider);
+  });
+
+  beforeEach(() => {
+    tracing = new OpenTelemetryTracer();
+  });
+
+  afterEach(() => {
+    span.end();
+  });
+
+  it('creates a span', () => {
+    span = tracing.createSpan(
+      spanName,
+      spanAttributes,
+      spanContext
+    ) as trace.Span;
+    assert.strictEqual(span.name, spanName);
+    assert.deepStrictEqual(span.attributes, spanAttributes);
+    assert.strictEqual(span.parentSpanId, spanContext.spanId);
+  });
+});