feat: Open telemetry integration and span Id fix for nodejs logging library

package.json

@@ -59,7 +59,8 @@
     "on-finished": "^2.3.0",
     "pumpify": "^2.0.1",
     "stream-events": "^1.0.5",
-    "uuid": "^9.0.0"
+    "uuid": "^9.0.0",
+    "@opentelemetry/api": "^1.7.0"
   },
   "devDependencies": {
     "@google-cloud/bigquery": "^7.0.0",
@@ -93,7 +94,14 @@
     "typescript": "^5.1.6",
     "uglify-js": "^3.13.5",
     "webpack": "^5.0.0",
-    "webpack-cli": "^5.0.0"
+    "webpack-cli": "^5.0.0",
+    "@opentelemetry/api": "^1.7.0",
+    "@opentelemetry/sdk-trace-node": "^1.23.0",
+    "@opentelemetry/sdk-node": "^0.50.0",
+    "@opentelemetry/semantic-conventions": "^1.23.0",
+    "@opentelemetry/sdk-trace-base": "^1.23.0",
+    "@opentelemetry/resources": "^1.23.0",
+    "@google-cloud/opentelemetry-cloud-trace-exporter": "^2.1.0"
   },
   "engines": {
     "node": ">=14.0.0"

src/entry.ts

@@ -30,7 +30,11 @@ import {
   RawHttpRequest,
   isRawHttpRequest,
 } from './utils/http-request';
-import {CloudTraceContext, getOrInjectContext} from './utils/context';
+import {
+  CloudTraceContext,
+  getContextFromOtelContext,
+  getOrInjectContext,
+} from './utils/context';
 
 const eventId = new EventId();
 
@@ -229,16 +233,17 @@ class Entry {
     const req = this.metadata.httpRequest;
     if (isRawHttpRequest(req)) {
       entry.httpRequest = makeHttpRequestData(req);
-      // Format trace and span
-      const traceContext = this.extractTraceFromHeaders(projectId);
-      if (traceContext) {
-        if (!this.metadata.trace && traceContext.trace)
-          entry.trace = traceContext.trace;
-        if (!this.metadata.spanId && traceContext.spanId)
-          entry.spanId = traceContext.spanId;
-        if (this.metadata.traceSampled === undefined)
-          entry.traceSampled = traceContext.traceSampled;
-      }
+    }
+
+    // Format trace and span
+    const traceContext = this.extractTraceContext(projectId);
+    if (traceContext) {
+      if (!this.metadata.trace && traceContext.trace)
+        entry.trace = traceContext.trace;
+      if (!this.metadata.spanId && traceContext.spanId)
+        entry.spanId = traceContext.spanId;
+      if (this.metadata.traceSampled === undefined)
+        entry.traceSampled = traceContext.traceSampled;
     }
     return entry;
   }
@@ -308,26 +313,31 @@ class Entry {
     const req = meta.httpRequest;
     if (isRawHttpRequest(req)) {
       entry.httpRequest = makeHttpRequestData(req);
-      // Detected trace context from headers if applicable.
-      const traceContext = this.extractTraceFromHeaders(projectId);
-      if (traceContext) {
-        if (!entry[TRACE_KEY] && traceContext.trace)
-          entry[TRACE_KEY] = traceContext.trace;
-        if (!entry[SPAN_ID_KEY] && traceContext.spanId)
-          entry[SPAN_ID_KEY] = traceContext.spanId;
-        if (entry[TRACE_SAMPLED_KEY] === undefined)
-          entry[TRACE_SAMPLED_KEY] = traceContext.traceSampled;
-      }
+    }
+
+    // Detected trace context from OpenTelemetry context or http headers if applicable.
+    const traceContext = this.extractTraceContext(projectId);
+    if (traceContext) {
+      if (!entry[TRACE_KEY] && traceContext.trace)
+        entry[TRACE_KEY] = traceContext.trace;
+      if (!entry[SPAN_ID_KEY] && traceContext.spanId)
+        entry[SPAN_ID_KEY] = traceContext.spanId;
+      if (entry[TRACE_SAMPLED_KEY] === undefined)
+        entry[TRACE_SAMPLED_KEY] = traceContext.traceSampled;
     }
     return entry;
   }
 
   /**
-   * extractTraceFromHeaders extracts trace and span information from raw HTTP
-   * request headers only.
+   * extractTraceContext extracts trace and span information from OpenTelemetry
+   * span context or raw HTTP request headers.
    * @private
    */
-  private extractTraceFromHeaders(projectId: string): CloudTraceContext | null {
+  private extractTraceContext(projectId: string): CloudTraceContext | null {
+    // Extract trace context from OpenTelemetry span context.
+    const otelContext = getContextFromOtelContext(projectId);
+    if (otelContext) return otelContext;
+    // Extract trace context from http request headers.
     const rawReq = this.metadata.httpRequest;
     if (rawReq && 'headers' in rawReq) {
       return getOrInjectContext(rawReq, projectId, false);

src/index.ts

@@ -1604,3 +1604,4 @@ module.exports.v2 = v2;
 import * as protos from '../protos/protos';
 export {protos};
 export {v2};
+export * from '@opentelemetry/api';

src/utils/context.ts

@@ -30,6 +30,7 @@
 import * as http from 'http';
 import * as uuid from 'uuid';
 import * as crypto from 'crypto';
+import {trace, isSpanContextValid} from '@opentelemetry/api';
 
 /** Header that carries span context across Google infrastructure. */
 export const X_CLOUD_TRACE_HEADER = 'x-cloud-trace-context';
@@ -98,7 +99,13 @@ export function getOrInjectContext(
   inject?: boolean
 ): CloudTraceContext {
   const defaultContext = toCloudTraceContext({}, projectId);
+
+  // Get trace context from OpenTelemetry span context.
+  const otelContext = getContextFromOtelContext(projectId);
+  if (otelContext) return otelContext;
+
   const wrapper = makeHeaderWrapper(req);
+
   if (wrapper) {
     // Detect 'traceparent' header.
     const traceContext = getContextFromTraceParent(wrapper, projectId);
@@ -149,6 +156,30 @@ function makeCloudTraceHeader(): string {
   return `${trace}/${spanId}`;
 }
 
+/**
+ * getContextFromOtelContext looks for the active open telemetry span context
+ * per Open Telemetry specifications for tracing contexts.
+ *
+ * @param projectId
+ */
+export function getContextFromOtelContext(
+  projectId: string
+): CloudTraceContext | null {
+  const spanContext = trace.getActiveSpan()?.spanContext();
+  const FLAG_SAMPLED = 1; // 00000001
+  if (spanContext !== undefined && isSpanContextValid(spanContext)) {
+    const otelSpanContext = {
+      trace: spanContext?.traceId,
+      spanId: spanContext?.spanId,
+      traceSampled: (spanContext.traceFlags & FLAG_SAMPLED) !== 0,
+    };
+
+    return toCloudTraceContext(otelSpanContext, projectId);
+  }
+
+  return null;
+}
+
 /**
  * getContextFromXCloudTrace looks for the HTTP header 'x-cloud-trace-context'
  * per Google Cloud specifications for Cloud Tracing.

system-test/logging.ts

@@ -29,8 +29,11 @@
 import {Logging, Sink, Log, Entry, TailEntriesResponse} from '../src';
 import * as http from 'http';
 import * as instrumentation from '../src/utils/instrumentation';
-
-// block all attempts to chat with the metadata server (kokoro runs on GCE)
+import {trace} from '@opentelemetry/api';
+import {Resource} from '@opentelemetry/resources';
+import {SEMRESATTRS_SERVICE_NAME} from '@opentelemetry/semantic-conventions';
+import {TraceExporter} from '@google-cloud/opentelemetry-cloud-trace-exporter';
+import {NodeSDK} from '@opentelemetry/sdk-node';
 nock(HOST_ADDRESS)
   .get(() => true)
   .replyWithError({code: 'ENOTFOUND'})
@@ -85,7 +88,7 @@
     // Fixes: https://github.com/googleapis/nodejs-logging/issues/953
     async function deleteBuckets() {
       const [buckets] = await storage.getBuckets({prefix: TESTS_PREFIX});
       const bucketsToDelete = buckets.filter((bucket: any) => {
         if (bucket.metadata.timeCreated) {
           return new Date(bucket.metadata.timeCreated) < twoDaysAgo;
         } else {
@@ -270,8 +273,8 @@
   });
 
   describe('logs', () => {
-    function getTestLog(loggingInstnce = null) {
-      const log = (loggingInstnce || logging).log(generateName());
+    function getTestLog(loggingInstance = null) {
+      const log = (loggingInstance || logging).log(generateName());
 
       const logEntries = [
         // string data
@@ -733,15 +736,185 @@
       });
     });
 
+    describe('logs with open telemetry context', () => {
+      let sdk: NodeSDK;
+      before(() => {
+        // initialize the SDK and register with the OpenTelemetry API
+        // this enables the API to record telemetry
+        sdk = new NodeSDK({
+          resource: new Resource({
+            [SEMRESATTRS_SERVICE_NAME]: TESTS_PREFIX,
+          }),
+          // Add cloud trace exporter as SDK trace exporter
+          traceExporter: new TraceExporter(),
+        });
+        sdk.start();
+      });
+
+      after(() => {
+        sdk.shutdown();
+      });
+
+      it('should not overwrite user defined trace and spans with OpenTelemetry context', done => {
+        trace.getTracer(TESTS_PREFIX).startActiveSpan('foo', span => {
+          const {log} = getTestLog();
+          const spanTestMessage = 'span test log message';
+          const metadata = {
+            trace: '1',
+            spanId: '1',
+            traceSampled: false,
+          };
+          const logEntry = log.entry(metadata, spanTestMessage);
+          log.write(logEntry, err => {
+            assert.ifError(err);
+            getEntriesFromLog(log, {numExpectedMessages: 1}, (err, entries) => {
+              assert.ifError(err);
+              const entry = entries![0];
+              assert.strictEqual(entry.data, spanTestMessage);
+              assert.strictEqual(entry.metadata.trace, metadata.trace);
+              assert.strictEqual(entry.metadata.spanId, metadata.spanId);
+              assert.strictEqual(
+                entry.metadata.traceSampled,
+                metadata.traceSampled
+              );
+            });
+          });
+          span.end();
+        });
+        done();
+      });
+
+      it('should write a log with trace and spans from OpenTelemetry context', done => {
+        trace.getTracer(TESTS_PREFIX).startActiveSpan('foo', span => {
+          const traceId = span.spanContext().traceId;
+          const spanId = span.spanContext().spanId;
+          const traceSampled = (span.spanContext().traceFlags & 1) !== 0;
+          const {log} = getTestLog();
+          const spanTestMessage = 'span test log message';
+          const logEntry = log.entry(spanTestMessage);
+          log.write(logEntry, err => {
+            assert.ifError(err);
+            getEntriesFromLog(log, {numExpectedMessages: 1}, (err, entries) => {
+              assert.ifError(err);
+              const entry = entries![0];
+              assert.strictEqual(entry.data, spanTestMessage);
+              assert.strictEqual(
+                entry.metadata.trace,
+                `projects/${PROJECT_ID}/traces/${traceId}`
+              );
+              assert.strictEqual(entry.metadata.spanId, spanId);
+              assert.strictEqual(entry.metadata.traceSampled, traceSampled);
+            });
+          });
+          span.end();
+        });
+        done();
+      });
+
+      it('should write a log with OpenTelemetry trace and spans and ignore http requests traceparent header', done => {
+        const {log} = getTestLog();
+        const URL = 'http://www.google.com';
+        trace.getTracer(TESTS_PREFIX).startActiveSpan('foo', span => {
+          // Use the response of a http request as the incomingmessage request obj.
+          http.get(URL, res => {
+            res.url = URL;
+            res.headers = {
+              traceparent:
+                '00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01',
+            };
+            const metadata = {httpRequest: res};
+            const logEntry = log.entry(metadata, 'some log message');
+
+            const traceId = span.spanContext().traceId;
+            const spanId = span.spanContext().spanId;
+            const traceSampled = (span.spanContext().traceFlags & 1) !== 0;
+
+            log.write(logEntry, err => {
+              assert.ifError(err);
+              getEntriesFromLog(
+                log,
+                {numExpectedMessages: 1},
+                (err, entries) => {
+                  assert.ifError(err);
+                  const entry = entries![0];
+                  assert.strictEqual(entry.data, 'some log message');
+                  assert.strictEqual(
+                    entry.metadata.httpRequest?.requestUrl,
+                    URL
+                  );
+                  assert.strictEqual(
+                    entry.metadata.httpRequest?.protocol,
+                    'http:'
+                  );
+                  assert.strictEqual(
+                    entry.metadata.trace,
+                    `projects/${PROJECT_ID}/traces/${traceId}`
+                  );
+                  assert.strictEqual(entry.metadata.spanId, spanId);
+                  assert.strictEqual(entry.metadata.traceSampled, traceSampled);
+                }
+              );
+            });
+          });
+          span.end();
+        });
+        done();
+      });
+
+      it('should write a log with OpenTelemetry trace and spans and ignore http requests x-cloud-trace-context header', done => {
+        const {log} = getTestLog();
+        const URL = 'http://www.google.com';
+        trace.getTracer(TESTS_PREFIX).startActiveSpan('foo', span => {
+          // Use the response of a http request as the incomingmessage request obj.
+          http.get(URL, res => {
+            res.url = URL;
+            res.headers = {
+              'x-cloud-trace-context': '1/2;o=1',
+            };
+            const metadata = {httpRequest: res};
+            const logEntry = log.entry(metadata, 'some log message');
+            const traceId = span.spanContext().traceId;
+            const spanId = span.spanContext().spanId;
+            const traceSampled = (span.spanContext().traceFlags & 1) !== 0;
+            log.write(logEntry, err => {
+              assert.ifError(err);
+              getEntriesFromLog(
+                log,
+                {numExpectedMessages: 1},
+                (err, entries) => {
+                  assert.ifError(err);
+                  const entry = entries![0];
+                  assert.strictEqual(entry.data, 'some log message');
+                  assert.strictEqual(
+                    entry.metadata.httpRequest?.requestUrl,
+                    URL
+                  );
+                  assert.strictEqual(
+                    entry.metadata.httpRequest?.protocol,
+                    'http:'
+                  );
+                  assert.strictEqual(
+                    entry.metadata.trace,
+                    `projects/${PROJECT_ID}/traces/${traceId}`
+                  );
+                  assert.strictEqual(entry.metadata.spanId, spanId);
+                  assert.strictEqual(entry.metadata.traceSampled, traceSampled);
+                }
+              );
+            });
+          });
+          span.end();
+        });
+        done();
+      });
+    });
+
     it('should set the default resource', done => {
       const {log} = getTestLog();
-
       const text = 'entry-text';
       const entry = log.entry(text);
-
       log.write(entry, err => {
         assert.ifError(err);
-
         getEntriesFromLog(log, {numExpectedMessages: 1}, (err, entries) => {
           assert.ifError(err);
           const entry = entries![0];