feat: (observability): trace Database.runPartitionedUpdate

observability-test/database.ts

@@ -38,7 +38,10 @@
   InMemorySpanExporter,
 } = require('@opentelemetry/sdk-trace-node');
 // eslint-disable-next-line n/no-extraneous-require
-const {SimpleSpanProcessor} = require('@opentelemetry/sdk-trace-base');
+const {
+  ReadableSpan,
+  SimpleSpanProcessor,
+} = require('@opentelemetry/sdk-trace-base');
 import * as db from '../src/database';
 import {Instance, MutationGroup, Spanner} from '../src';
 import * as pfy from '@google-cloud/promisify';
@@ -90,7 +93,7 @@
 
 class FakeSession {
   calledWith_: IArguments;
   formattedName_: any;
   constructor() {
     this.calledWith_ = arguments;
   }
@@ -504,7 +507,7 @@
 
     let beginSnapshotStub: sinon.SinonStub;
     let getSessionStub: sinon.SinonStub;
     let snapshotStub: sinon.SinonStub;
 
     beforeEach(() => {
       fakePool = database.pool_;
@@ -1172,7 +1175,7 @@
 
     it('with error on null mutation should catch thrown error', done => {
       try {
         database.writeAtLeastOnce(null, (err, res) => {});
       } catch (err) {
         // Performing a substring search on the error because
         // depending on the version of Node.js, the error might be either of:
@@ -1954,4 +1957,220 @@
       fakeStream2.push(null);
     });
   });
+
+  describe('runPartitionedUpdate', () => {
+    const QUERY = {
+      sql: 'INSERT INTO `MyTable` (Key, Thing) VALUES(@key, @thing)',
+      params: {
+        key: 'k999',
+        thing: 'abc',
+      },
+    };
+
+    let fakePool: FakeSessionPool;
+    let fakeSession: FakeSession;
+    let fakePartitionedDml = new FakeTransaction(
+      {} as google.spanner.v1.TransactionOptions.PartitionedDml
+    );
+
+    let getSessionStub;
+    let beginStub;
+    let runUpdateStub;
+
+    beforeEach(() => {
+      fakePool = database.pool_;
+      fakeSession = new FakeSession();
+      fakePartitionedDml = new FakeTransaction(
+        {} as google.spanner.v1.TransactionOptions.PartitionedDml
+      );
+
+      getSessionStub = (
+        sandbox.stub(fakePool, 'getSession') as sinon.SinonStub
+      ).callsFake(callback => {
+        callback(null, fakeSession);
+      });
+
+      sandbox.stub(fakeSession, 'partitionedDml').returns(fakePartitionedDml);
+
+      beginStub = (
+        sandbox.stub(fakePartitionedDml, 'begin') as sinon.SinonStub
+      ).callsFake(callback => callback(null));
+
+      runUpdateStub = (
+        sandbox.stub(fakePartitionedDml, 'runUpdate') as sinon.SinonStub
+      ).callsFake((_, callback) => callback(null));
+    });
+
+    interface traceExportResults {
+      spanNames: string[];
+      spans: (typeof ReadableSpan)[];
+      eventNames: string[];
+    }
+
+    async function getTraceExportResults(): Promise<traceExportResults> {
+      await provider.forceFlush();
+      await traceExporter.forceFlush();
+      const spans = traceExporter.getFinishedSpans();
+      withAllSpansHaveDBName(spans);
+
+      const actualSpanNames: string[] = [];
+      const actualEventNames: string[] = [];
+      spans.forEach(span => {
+        actualSpanNames.push(span.name);
+        span.events.forEach(event => {
+          actualEventNames.push(event.name);
+        });
+      });
+
+      return Promise.resolve({
+        spanNames: actualSpanNames,
+        spans: spans,
+        eventNames: actualEventNames,
+      });
+    }
+
+    it('with pool errors', done => {
+      const fakeError = new Error('err');
+      const fakeCallback = sandbox.spy();
+
+      getSessionStub.callsFake(callback => callback(fakeError));
+      database.runPartitionedUpdate(QUERY, async (err, rowCount) => {
+        assert.strictEqual(err, fakeError);
+        assert.strictEqual(rowCount, 0);
+
+        const exportResults = await getTraceExportResults();
+        const actualSpanNames = exportResults.spanNames;
+        const spans = exportResults.spans;
+        const actualEventNames = exportResults.eventNames;
+
+        const expectedSpanNames = [
+          'CloudSpanner.Database.runPartitionedUpdate',
+        ];
+        assert.deepStrictEqual(
+          actualSpanNames,
+          expectedSpanNames,
+          `span names mismatch:\n\tGot:  ${actualSpanNames}\n\tWant: ${expectedSpanNames}`
+        );
+
+        // Ensure that the first span actually produced an error that was recorded.
+        const parentSpan = spans[0];
+        assert.deepStrictEqual(
+          SpanStatusCode.ERROR,
+          parentSpan.status.code,
+          'Expected an ERROR span status'
+        );
+        assert.deepStrictEqual(
+          fakeError.message,
+          parentSpan.status.message.toString(),
+          'Mismatched span status message'
+        );
+
+        const expectedEventNames = [];
+        assert.deepStrictEqual(
+          actualEventNames,
+          expectedEventNames,
+          `Unexpected events:\n\tGot:  ${actualEventNames}\n\tWant: ${expectedEventNames}`
+        );
+
+        done();
+      });
+    });
+
+    it('with begin errors', done => {
+      const fakeError = new Error('err');
+
+      beginStub.callsFake(callback => callback(fakeError));
+
+      const releaseStub = (
+        sandbox.stub(fakePool, 'release') as sinon.SinonStub
+      ).withArgs(fakeSession);
+
+      database.runPartitionedUpdate(QUERY, async (err, rowCount) => {
+        assert.strictEqual(err, fakeError);
+        assert.strictEqual(rowCount, 0);
+        assert.strictEqual(releaseStub.callCount, 1);
+
+        const exportResults = await getTraceExportResults();
+        const actualSpanNames = exportResults.spanNames;
+        const spans = exportResults.spans;
+        const actualEventNames = exportResults.eventNames;
+
+        const expectedSpanNames = [
+          'CloudSpanner.Database.runPartitionedUpdate',
+        ];
+        assert.deepStrictEqual(
+          actualSpanNames,
+          expectedSpanNames,
+          `span names mismatch:\n\tGot:  ${actualSpanNames}\n\tWant: ${expectedSpanNames}`
+        );
+
+        // Ensure that the first span actually produced an error that was recorded.
+        const parentSpan = spans[0];
+        assert.deepStrictEqual(
+          SpanStatusCode.ERROR,
+          parentSpan.status.code,
+          'Expected an ERROR span status'
+        );
+        assert.deepStrictEqual(
+          fakeError.message,
+          parentSpan.status.message.toString(),
+          'Mismatched span status message'
+        );
+
+        const expectedEventNames = [];
+        assert.deepStrictEqual(
+          actualEventNames,
+          expectedEventNames,
+          `Unexpected events:\n\tGot:  ${actualEventNames}\n\tWant: ${expectedEventNames}`
+        );
+        done();
+      });
+    });
+
+    it('session released on transaction end', done => {
+      const releaseStub = (
+        sandbox.stub(fakePool, 'release') as sinon.SinonStub
+      ).withArgs(fakeSession);
+
+      database.runPartitionedUpdate(QUERY, async (err, rowCount) => {
+        const exportResults = await getTraceExportResults();
+        const actualSpanNames = exportResults.spanNames;
+        const spans = exportResults.spans;
+        const actualEventNames = exportResults.eventNames;
+
+        const expectedSpanNames = [
+          'CloudSpanner.Database.runPartitionedUpdate',
+        ];
+        assert.deepStrictEqual(
+          actualSpanNames,
+          expectedSpanNames,
+          `span names mismatch:\n\tGot:  ${actualSpanNames}\n\tWant: ${expectedSpanNames}`
+        );
+
+        // Ensure that the first span actually produced an error that was recorded.
+        const parentSpan = spans[0];
+        assert.deepStrictEqual(
+          SpanStatusCode.UNSET,
+          parentSpan.status.code,
+          'Unexpected span status'
+        );
+        assert.deepStrictEqual(
+          undefined,
+          parentSpan.status.message,
+          'Mismatched span status message'
+        );
+
+        const expectedEventNames = [];
+        assert.deepStrictEqual(
+          actualEventNames,
+          expectedEventNames,
+          `Unexpected events:\n\tGot:  ${actualEventNames}\n\tWant: ${expectedEventNames}`
+        );
+        done();
+      });
+
+      fakePartitionedDml.emit('end');
+      assert.strictEqual(releaseStub.callCount, 1);
+    });
+  });
 });

observability-test/spanner.ts

@@ -613,6 +613,53 @@ describe('EndToEnd', async () => {
         done();
       });
     });
+
+    it('runPartitionedUpdate', async () => {
+      const [rowCount] = await database.runPartitionedUpdate({
+        sql: updateSql,
+      });
+
+      await tracerProvider.forceFlush();
+      await traceExporter.forceFlush();
+      const spans = traceExporter.getFinishedSpans();
+
+      const actualEventNames: string[] = [];
+      const actualSpanNames: string[] = [];
+      spans.forEach(span => {
+        actualSpanNames.push(span.name);
+        span.events.forEach(event => {
+          actualEventNames.push(event.name);
+        });
+      });
+
+      const expectedSpanNames = [
+        'CloudSpanner.Snapshot.begin',
+        'CloudSpanner.Snapshot.runStream',
+        'CloudSpanner.Snapshot.run',
+        'CloudSpanner.Dml.runUpdate',
+        'CloudSpanner.PartitionedDml.runUpdate',
+        'CloudSpanner.Database.runPartitionedUpdate',
+      ];
+      assert.deepStrictEqual(
+        actualSpanNames,
+        expectedSpanNames,
+        `span names mismatch:\n\tGot:  ${actualSpanNames}\n\tWant: ${expectedSpanNames}`
+      );
+
+      const expectedEventNames = [
+        'Begin Transaction',
+        'Transaction Creation Done',
+        'Starting stream',
+        'Acquiring session',
+        'Cache hit: has usable session',
+        'Acquired session',
+      ];
+      assert.deepStrictEqual(
+        actualEventNames,
+        expectedEventNames,
+        `Unexpected events:\n\tGot:  ${actualEventNames}\n\tWant: ${expectedEventNames}`
+      );
+    });
   });
 });
 

src/database.ts

@@ -2858,13 +2858,27 @@ class Database extends common.GrpcServiceObject {
     query: string | RunPartitionedUpdateOptions,
     callback?: RunUpdateCallback
   ): void | Promise<[number]> {
-    this.pool_.getSession((err, session) => {
-      if (err) {
-        callback!(err as ServiceError, 0);
-        return;
-      }
+    const traceConfig = {
+      sql: query,
+      ...this._traceConfig,
+    };
+    return startTrace('Database.runPartitionedUpdate', traceConfig, span => {
+      this.pool_.getSession((err, session) => {
+        if (err) {
+          setSpanError(span, err);
+          span.end();
+          callback!(err as ServiceError, 0);
+          return;
+        }
 
-      this._runPartitionedUpdate(session!, query, callback);
+        this._runPartitionedUpdate(session!, query, (err, count) => {
+          if (err) {
+            setSpanError(span, err);
+          }
+          span.end();
+          callback!(err, count);
+        });
+      });
     });
   }