Introduce API Metrics/Telemetry (#1142)

Neo4j would like to be able to track more information about driver apis usage, so that smarter decisions can be made on improving drivers and their stack.

The collection of metrics must be respectful of users. Thus, the collected metrics are:
1. impossible to be tied back the customer or user
1. transparent to the users that inspect
1. restrained to required metrics

The metric collection is disabled by default in Neo4j. It can be enabled in the server by setting `server.bolt.telemetry.enabled` to `true`.

However, the metric collection is enabled by default in the drivers. It can be disabled in the driver by configuring the driver with `telemetryDisabled` to `true`. Default: `false`.

**Metrics are only collected when enabled both in server and driver instances.**

Disabling metrics on driver:
```typescript
const driver = neo4j.driver(
    'neo4j://localhost:7687', neo4j.auth.basic('neo4j', 'password'), { telemetryDisabled: true })
```

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

Author: bigmontz

packages/bolt-connection/src/bolt/bolt-protocol-v1.js

@@ -27,6 +27,7 @@ import { Chunker } from '../channel'
 import { structure, v1 } from '../packstream'
 import RequestMessage, { SIGNATURES } from './request-message'
 import {
+  CompletedObserver,
   LoginObserver,
   LogoffObserver,
   ResetObserver,
@@ -454,6 +455,24 @@ export default class BoltProtocol {
     return observer
   }
 
+  /**
+   * Send a TELEMETRY through the underlying connection.
+   *
+   * @param {object} param0 Message params
+   * @param {number} param0.api The API called
+   * @param {object} param1 Configuration and callbacks
+   * @param {function()} param1.onCompleted Called when completed
+   * @param {function()} param1.onError Called when error
+   * @return {StreamObserver} the stream observer that monitors the corresponding server response.
+   */
+  telemetry ({ api }, { onError, onCompleted } = {}) {
+    const observer = new CompletedObserver()
+    if (onCompleted) {
+      onCompleted()
+    }
+    return observer
+  }
+
   _createPacker (chunker) {
     return new v1.Packer(chunker)
   }

packages/bolt-connection/src/bolt/bolt-protocol-v5x4.js

@@ -0,0 +1,61 @@
+/**
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * 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 BoltProtocolV5x3 from './bolt-protocol-v5x3'
+
+import transformersFactories from './bolt-protocol-v5x4.transformer'
+import RequestMessage from './request-message'
+import { TelemetryObserver } from './stream-observers'
+import Transformer from './transformer'
+
+import { internal } from 'neo4j-driver-core'
+
+const {
+  constants: { BOLT_PROTOCOL_V5_4 }
+} = internal
+
+export default class BoltProtocol extends BoltProtocolV5x3 {
+  get version () {
+    return BOLT_PROTOCOL_V5_4
+  }
+
+  get transformer () {
+    if (this._transformer === undefined) {
+      this._transformer = new Transformer(Object.values(transformersFactories).map(create => create(this._config, this._log)))
+    }
+    return this._transformer
+  }
+
+  /**
+   * Send a TELEMETRY through the underlying connection.
+   *
+   * @param {object} param0 Message params
+   * @param {number} param0.api The API called
+   * @param {object} param1 Configuration and callbacks callbacks
+   * @param {function()} param1.onCompleted Called when completed
+   * @param {function()} param1.onError Called when error
+   * @return {StreamObserver} the stream observer that monitors the corresponding server response.
+   */
+  telemetry ({ api }, { onError, onCompleted } = {}) {
+    const observer = new TelemetryObserver({ onCompleted, onError })
+
+    this.write(RequestMessage.telemetry({ api }), observer, false)
+
+    return observer
+  }
+}

packages/bolt-connection/src/bolt/bolt-protocol-v5x4.transformer.js

@@ -0,0 +1,24 @@
+/**
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * 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 v5x3 from './bolt-protocol-v5x3.transformer'
+
+export default {
+  ...v5x3
+}

packages/bolt-connection/src/bolt/create.js

@@ -30,6 +30,7 @@ import BoltProtocolV5x0 from './bolt-protocol-v5x0'
 import BoltProtocolV5x1 from './bolt-protocol-v5x1'
 import BoltProtocolV5x2 from './bolt-protocol-v5x2'
 import BoltProtocolV5x3 from './bolt-protocol-v5x3'
+import BoltProtocolV5x4 from './bolt-protocol-v5x4'
 // eslint-disable-next-line no-unused-vars
 import { Chunker, Dechunker } from '../channel'
 import ResponseHandler from './response-handler'
@@ -222,6 +223,14 @@ function createProtocol (
         log,
         onProtocolError,
         serversideRouting)
+    case 5.4:
+      return new BoltProtocolV5x4(server,
+        chunker,
+        packingConfig,
+        createResponseHandler,
+        log,
+        onProtocolError,
+        serversideRouting)
     default:
       throw newError('Unknown Bolt protocol version: ' + version)
   }

packages/bolt-connection/src/bolt/handshake.js

@@ -78,7 +78,7 @@ function parseNegotiatedResponse (buffer, log) {
  */
 function newHandshakeBuffer () {
   return createHandshakeMessage([
-    [version(5, 3), version(5, 0)],
+    [version(5, 4), version(5, 0)],
     [version(4, 4), version(4, 2)],
     version(4, 1),
     version(3, 0)

packages/bolt-connection/src/bolt/request-message.js

@@ -38,6 +38,9 @@ const GOODBYE = 0x02 // 0000 0010 // GOODBYE
 const BEGIN = 0x11 // 0001 0001 // BEGIN <metadata>
 const COMMIT = 0x12 // 0001 0010 // COMMIT
 const ROLLBACK = 0x13 // 0001 0011 // ROLLBACK
+
+const TELEMETRY = 0x54 // 0101 0100 // TELEMETRY <api>
+
 const ROUTE = 0x66 // 0110 0110 // ROUTE
 
 const LOGON = 0x6A // LOGON
@@ -61,6 +64,7 @@ const SIGNATURES = Object.freeze({
   BEGIN,
   COMMIT,
   ROLLBACK,
+  TELEMETRY,
   ROUTE,
   LOGON,
   LOGOFF,
@@ -367,6 +371,15 @@ export default class RequestMessage {
     )
   }
 
+  static telemetry ({ api }) {
+    const parsedApi = int(api)
+    return new RequestMessage(
+      TELEMETRY,
+      [parsedApi],
+      () => `TELEMETRY ${parsedApi.toString()}`
+    )
+  }
+
   /**
    * Generate the ROUTE message, this message is used to fetch the routing table from the server
    *

packages/bolt-connection/src/bolt/stream-observers.js

@@ -526,6 +526,38 @@ class ResetObserver extends StreamObserver {
   }
 }
 
+class TelemetryObserver extends ResultStreamObserver {
+  /**
+   *
+   * @param {Object} param -
+   * @param {function(err: Error)} param.onError
+   * @param {function(metadata)} param.onCompleted
+   */
+  constructor ({ onError, onCompleted } = {}) {
+    super()
+    this._onError = onError
+    this._onCompleted = onCompleted
+  }
+
+  onNext (record) {
+    this.onError(
+      newError('Received RECORD when sending telemetry ' + json.stringify(record), PROTOCOL_ERROR)
+    )
+  }
+
+  onError (error) {
+    if (this._onError) {
+      this._onError(error)
+    }
+  }
+
+  onCompleted (metadata) {
+    if (this._onCompleted) {
+      this._onCompleted(metadata)
+    }
+  }
+}
+
 class FailedObserver extends ResultStreamObserver {
   constructor ({ error, onError }) {
     super({ beforeError: onError })
@@ -708,5 +740,6 @@ export {
   FailedObserver,
   CompletedObserver,
   RouteObserver,
-  ProcedureRouteObserver
+  ProcedureRouteObserver,
+  TelemetryObserver
 }

packages/bolt-connection/src/connection/connection-channel.js

@@ -87,7 +87,8 @@ export function createChannelConnection (
         serversideRouting,
         chunker,
         config.notificationFilter,
-        createProtocol
+        createProtocol,
+        config.telemetryDisabled
       )
 
       // forward all pending bytes to the dechunker
@@ -121,7 +122,8 @@ export default class ChannelConnection extends Connection {
     serversideRouting = null,
     chunker, // to be removed,
     notificationFilter,
-    protocolSupplier
+    protocolSupplier,
+    telemetryDisabled
   ) {
     super(errorHandler)
     this._authToken = null
@@ -137,6 +139,8 @@ export default class ChannelConnection extends Connection {
     this._log = createConnectionLogger(this, log)
     this._serversideRouting = serversideRouting
     this._notificationFilter = notificationFilter
+    this._telemetryDisabledDriverConfig = telemetryDisabled === true
+    this._telemetryDisabledConnection = true
 
     // connection from the database, returned in response for HELLO message and might not be available
     this._dbConnectionId = null
@@ -157,13 +161,31 @@ export default class ChannelConnection extends Connection {
   }
 
   beginTransaction (config) {
+    this._sendTelemetryIfEnabled(config)
     return this._protocol.beginTransaction(config)
   }
 
   run (query, parameters, config) {
+    this._sendTelemetryIfEnabled(config)
     return this._protocol.run(query, parameters, config)
   }
 
+  _sendTelemetryIfEnabled (config) {
+    if (this._telemetryDisabledConnection ||
+        this._telemetryDisabledDriverConfig ||
+        config == null ||
+        config.apiTelemetryConfig == null) {
+      return
+    }
+
+    this._protocol.telemetry({
+      api: config.apiTelemetryConfig.api
+    }, {
+      onCompleted: config.apiTelemetryConfig.onTelemetrySuccess,
+      onError: config.beforeError
+    })
+  }
+
   commitTransaction (config) {
     return this._protocol.commitTransaction(config)
   }
@@ -290,6 +312,11 @@ export default class ChannelConnection extends Connection {
                   )
                 }
               }
+
+              const telemetryEnabledHint = metadata.hints['telemetry.enabled']
+              if (telemetryEnabledHint === true) {
+                this._telemetryDisabledConnection = false
+              }
             }
           }
           resolve(self)

packages/bolt-connection/test/bolt/__snapshots__/bolt-protocol-v1.test.js.snap

@@ -1,12 +1,12 @@
 // Jest Snapshot v1, https://goo.gl/fbAQLP
 
-exports[`#unit BoltProtocolV1 .packable() should pack not pack graph types (Node) 1`] = `"It is not allowed to pass nodes in query parameters, given: (c:a {a:"b"})"`;
+exports[`#unit BoltProtocolV1 .packable() should resultant function not pack graph types (Node) 1`] = `"It is not allowed to pass nodes in query parameters, given: (c:a {a:"b"})"`;
 
-exports[`#unit BoltProtocolV1 .packable() should pack not pack graph types (Path) 1`] = `"It is not allowed to pass paths in query parameters, given: [object Object]"`;
+exports[`#unit BoltProtocolV1 .packable() should resultant function not pack graph types (Path) 1`] = `"It is not allowed to pass paths in query parameters, given: [object Object]"`;
 
-exports[`#unit BoltProtocolV1 .packable() should pack not pack graph types (Relationship) 1`] = `"It is not allowed to pass relationships in query parameters, given: (e)-[:a {b:"c"}]->(f)"`;
+exports[`#unit BoltProtocolV1 .packable() should resultant function not pack graph types (Relationship) 1`] = `"It is not allowed to pass relationships in query parameters, given: (e)-[:a {b:"c"}]->(f)"`;
 
-exports[`#unit BoltProtocolV1 .packable() should pack not pack graph types (UnboundRelationship) 1`] = `"It is not allowed to pass unbound relationships in query parameters, given: -[:a {b:"c"}]->"`;
+exports[`#unit BoltProtocolV1 .packable() should resultant function not pack graph types (UnboundRelationship) 1`] = `"It is not allowed to pass unbound relationships in query parameters, given: -[:a {b:"c"}]->"`;
 
 exports[`#unit BoltProtocolV1 .packable() should pack types introduced afterwards as Map (Date) 1`] = `
 {

packages/bolt-connection/test/bolt/__snapshots__/bolt-protocol-v2.test.js.snap

@@ -1,12 +1,12 @@
 // Jest Snapshot v1, https://goo.gl/fbAQLP
 
-exports[`#unit BoltProtocolV2 .packable() should pack not pack graph types (Node) 1`] = `"It is not allowed to pass nodes in query parameters, given: (c:a {a:"b"})"`;
+exports[`#unit BoltProtocolV2 .packable() should resultant function not pack graph types (Node) 1`] = `"It is not allowed to pass nodes in query parameters, given: (c:a {a:"b"})"`;
 
-exports[`#unit BoltProtocolV2 .packable() should pack not pack graph types (Path) 1`] = `"It is not allowed to pass paths in query parameters, given: [object Object]"`;
+exports[`#unit BoltProtocolV2 .packable() should resultant function not pack graph types (Path) 1`] = `"It is not allowed to pass paths in query parameters, given: [object Object]"`;
 
-exports[`#unit BoltProtocolV2 .packable() should pack not pack graph types (Relationship) 1`] = `"It is not allowed to pass relationships in query parameters, given: (e)-[:a {b:"c"}]->(f)"`;
+exports[`#unit BoltProtocolV2 .packable() should resultant function not pack graph types (Relationship) 1`] = `"It is not allowed to pass relationships in query parameters, given: (e)-[:a {b:"c"}]->(f)"`;
 
-exports[`#unit BoltProtocolV2 .packable() should pack not pack graph types (UnboundRelationship) 1`] = `"It is not allowed to pass unbound relationships in query parameters, given: -[:a {b:"c"}]->"`;
+exports[`#unit BoltProtocolV2 .packable() should resultant function not pack graph types (UnboundRelationship) 1`] = `"It is not allowed to pass unbound relationships in query parameters, given: -[:a {b:"c"}]->"`;
 
 exports[`#unit BoltProtocolV2 .unpack() should not unpack with wrong size (Date with less fields) 1`] = `"Wrong struct size for Date, expected 1 but was 0"`;