Instrument append and subscribe methods

.github/workflows/build_and_lint.yml

@@ -35,6 +35,8 @@ jobs:
 
     steps:
       - uses: actions/checkout@v3
+        with:
+          node-version: "14.x"
       - name: Install
         run: yarn
       - name: Build

.github/workflows/tests.yml

@@ -38,6 +38,9 @@ jobs:
 
           - name: streams
             path: ./src/streams
+
+          - name: opentelemetry
+            path: ./src/opentelemetry
     env:
       # Github only passes secrets to the main repo, so we need to skip some things if they are unavailable
       SECRETS_AVAILABLE: ${{ secrets.eventstore_cloud_id != null }}

README.md

@@ -8,10 +8,11 @@ EventStoreDB is the event-native database, where business events are immutably s
 
 This monorepo contains the following packages:
 
-| Subfolder                                    | Package                                                                     |
-| -------------------------------------------- | --------------------------------------------------------------------------- |
-| [`packages/db-client/`](packages/db-client/) | [`@eventstore/client`](https://www.npmjs.com/package/@eventstore/db-client) |
-| [`packages/test/`](packages/test/)           | Client internal tests                                                       |
+| Subfolder                                            | Package                                                                                |
+| ---------------------------------------------------- | -------------------------------------------------------------------------------------- |
+| [`packages/db-client/`](packages/db-client/)         | [`@eventstore/client`](https://www.npmjs.com/package/@eventstore/db-client)            |
+| [`packages/opentelemetry/`](packages/opentelemetry/) | [`@eventstore/opentelemetry`](https://www.npmjs.com/package/@eventstore/opentelemetry) |
+| [`packages/test/`](packages/test/)                   | Client internal tests                                                                  |
 
 ## Installation
 

lerna.json

@@ -1,5 +1,5 @@
-{
-  "version": "independent",
-  "npmClient": "yarn",
-  "useWorkspaces": true
-}
+{
+  "version": "independent",
+  "npmClient": "yarn",
+  "useWorkspaces": true
+}

package.json

@@ -4,14 +4,15 @@
   "scripts": {
     "build": "lerna run --stream build",
     "build:ts": "lerna run --stream build:ts",
-    "build:watch": "lerna run --stream build:watch",
+    "build:watch": "lerna run build:watch --parallel",
     "lint": "run-s -c lint:*",
     "lint:prettier": "prettier --check \"packages/*/src/**/**/!(*.d).{ts,json}\"",
     "lint:eslint": "eslint \"packages/*/src/**/*.ts\"",
     "prettier:fix": "prettier --write \"packages/*/src/**/**/!(*.d).{ts,json}\"",
     "clean": "lerna run --stream clean",
     "prepublishOnly": "run-s clean build test",
-    "test": "lerna run --stream test --"
+    "test": "lerna run --stream test --",
+    "postinstall": "yarn build"
   },
   "author": "Event Store Limited",
   "license": "Apache-2.0",
@@ -27,6 +28,7 @@
     "@types/node": "^16.18.67",
     "@typescript-eslint/eslint-plugin": "^5.62.0",
     "@typescript-eslint/parser": "^5.62.0",
+    "cross-env": "^7.0.3",
     "eslint": "^7.32.0",
     "eslint-plugin-jsdoc": "^40.3.0",
     "lerna": "^6.6.2",

packages/db-client/src/Client/index.ts

@@ -594,7 +594,7 @@ export class Client {
     });
   };
 
-  private resolveUri = async (): Promise<string> => {
+  protected resolveUri = async (): Promise<string> => {
     if (this.#nextChannelSettings?.nextEndpoint) {
       const { address, port } = this.#nextChannelSettings.nextEndpoint;
       return `${address}:${port}`;

packages/db-client/src/persistentSubscription/utils/PersistentSubscriptionImpl.ts

@@ -24,7 +24,8 @@ export class PersistentSubscriptionImpl<E>
   implements PersistentSubscriptionBase<E>
 {
   #grpcStream: Promise<ClientDuplexStream<ReadReq, ReadResp>>;
-  #convertGrpcEvent: ConvertGrpcEvent<ReadResp.ReadEvent, E>;
+  protected convertGrpcEvent: ConvertGrpcEvent<ReadResp.ReadEvent, E>;
+  public id?: string;
 
   constructor(
     createGRPCStream: CreateGRPCStream,
@@ -33,7 +34,7 @@ export class PersistentSubscriptionImpl<E>
   ) {
     super({ ...options, objectMode: true });
     this.#grpcStream = createGRPCStream();
-    this.#convertGrpcEvent = convertGrpcEvent;
+    this.convertGrpcEvent = convertGrpcEvent;
     this.initialize();
   }
 
@@ -53,11 +54,12 @@ export class PersistentSubscriptionImpl<E>
 
   _transform(resp: ReadResp, _encoding: string, next: TransformCallback): void {
     if (resp.hasSubscriptionConfirmation()) {
+      this.id = resp.getSubscriptionConfirmation()?.getSubscriptionId();
       this.emit("confirmation");
     }
 
     if (resp.hasEvent()) {
-      const resolved = this.#convertGrpcEvent(resp.getEvent()!);
+      const resolved = this.convertGrpcEvent(resp.getEvent()!);
       next(null, resolved);
       return;
     }

packages/db-client/src/streams/utils/Subscription.ts

@@ -17,9 +17,10 @@ export class Subscription<E>
   extends Transform
   implements ReadableSubscription<E>
 {
-  #convertGrpcEvent: ConvertGrpcEvent<ReadResp.ReadEvent, E>;
+  protected convertGrpcEvent: ConvertGrpcEvent<ReadResp.ReadEvent, E>;
   #grpcStream: Promise<ClientReadableStream<ReadResp>>;
   #checkpointReached?: Filter["checkpointReached"];
+  public id?: string;
 
   constructor(
     createGRPCStream: CreateGRPCStream,
@@ -28,7 +29,7 @@ export class Subscription<E>
     checkpointReached?: Filter["checkpointReached"]
   ) {
     super({ ...options, objectMode: true });
-    this.#convertGrpcEvent = convertGrpcEvent;
+    this.convertGrpcEvent = convertGrpcEvent;
     this.#grpcStream = createGRPCStream();
     this.#checkpointReached = checkpointReached;
     this.initialize();
@@ -54,6 +55,7 @@ export class Subscription<E>
     next: TransformCallback
   ): Promise<void> {
     if (resp.hasConfirmation?.()) {
+      this.id = resp.getConfirmation()?.getSubscriptionId();
       this.emit("confirmation");
     }
 
@@ -75,7 +77,7 @@ export class Subscription<E>
     }
 
     if (resp.hasEvent?.()) {
-      const resolved = this.#convertGrpcEvent(resp.getEvent()!);
+      const resolved = this.convertGrpcEvent(resp.getEvent()!);
       return next(null, resolved);
     }
 

packages/db-client/src/types/index.ts

@@ -498,16 +498,26 @@ export interface CatchupSubscription {
   removeListener(event: "fellBehind", listener: () => void): this;
 }
 
+export interface SubscriptionDetails {
+  id?: string;
+}
+
 export type PersistentSubscriptionToStream<E extends EventType = EventType> =
-  PersistentSubscriptionBase<PersistentSubscriptionToStreamResolvedEvent<E>>;
+  PersistentSubscriptionBase<PersistentSubscriptionToStreamResolvedEvent<E>> &
+    SubscriptionDetails;
 
 export type PersistentSubscriptionToAll =
-  PersistentSubscriptionBase<PersistentSubscriptionToAllResolvedEvent>;
+  PersistentSubscriptionBase<PersistentSubscriptionToAllResolvedEvent> &
+    SubscriptionDetails;
 
 export type StreamSubscription<E extends EventType = EventType> =
-  ReadableSubscription<ResolvedEvent<E>> & CatchupSubscription;
+  ReadableSubscription<ResolvedEvent<E>> &
+    CatchupSubscription &
+    SubscriptionDetails;
 export type AllStreamSubscription =
-  ReadableSubscription<AllStreamResolvedEvent> & CatchupSubscription;
+  ReadableSubscription<AllStreamResolvedEvent> &
+    CatchupSubscription &
+    SubscriptionDetails;
 
 export { VNodeState };
 export * from "./events";

packages/opentelemetry/.gitignore

@@ -0,0 +1,155 @@
+### Node ###
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+lerna-debug.log*
+.pnpm-debug.log*
+
+# Diagnostic reports (https://nodejs.org/api/report.html)
+report.[0-9]*.[0-9]*.[0-9]*.[0-9]*.json
+
+# Runtime data
+pids
+*.pid
+*.seed
+*.pid.lock
+
+# Directory for instrumented libs generated by jscoverage/JSCover
+lib-cov
+
+# Coverage directory used by tools like istanbul
+coverage
+*.lcov
+
+# nyc test coverage
+.nyc_output
+
+# Grunt intermediate storage (https://gruntjs.com/creating-plugins#storing-task-files)
+.grunt
+
+# Bower dependency directory (https://bower.io/)
+bower_components
+
+# node-waf configuration
+.lock-wscript
+
+# Compiled binary addons (https://nodejs.org/api/addons.html)
+build/Release
+
+# Dependency directories
+node_modules/
+jspm_packages/
+
+# Snowpack dependency directory (https://snowpack.dev/)
+web_modules/
+
+# TypeScript cache
+*.tsbuildinfo
+
+# Optional npm cache directory
+.npm
+
+# Optional eslint cache
+.eslintcache
+
+# Optional stylelint cache
+.stylelintcache
+
+# Microbundle cache
+.rpt2_cache/
+.rts2_cache_cjs/
+.rts2_cache_es/
+.rts2_cache_umd/
+
+# Optional REPL history
+.node_repl_history
+
+# Output of 'npm pack'
+*.tgz
+
+# Yarn Integrity file
+.yarn-integrity
+
+# dotenv environment variable files
+.env
+.env.development.local
+.env.test.local
+.env.production.local
+.env.local
+
+# parcel-bundler cache (https://parceljs.org/)
+.cache
+.parcel-cache
+
+# Next.js build output
+.next
+out
+
+# Nuxt.js build / generate output
+.nuxt
+dist
+
+# Gatsby files
+.cache/
+# Comment in the public line in if your project uses Gatsby and not Next.js
+# https://nextjs.org/blog/next-9-1#public-directory-support
+# public
+
+# vuepress build output
+.vuepress/dist
+
+# vuepress v2.x temp and cache directory
+.temp
+
+# Docusaurus cache and generated files
+.docusaurus
+
+# Serverless directories
+.serverless/
+
+# FuseBox cache
+.fusebox/
+
+# DynamoDB Local files
+.dynamodb/
+
+# TernJS port file
+.tern-port
+
+# Stores VSCode versions used for testing VSCode extensions
+.vscode-test
+
+# yarn v2
+.yarn/cache
+.yarn/unplugged
+.yarn/build-state.yml
+.yarn/install-state.gz
+.pnp.*
+
+### Node Patch ###
+# Serverless Webpack directories
+.webpack/
+
+# Optional stylelint cache
+
+# SvelteKit build / generate output
+.svelte-kit
+
+### yarn ###
+# https://yarnpkg.com/getting-started/qa#which-files-should-be-gitignored
+
+.yarn/*
+!.yarn/releases
+!.yarn/patches
+!.yarn/plugins
+!.yarn/sdks
+!.yarn/versions
+
+# if you are NOT using Zero-installs, then:
+# comment the following lines
+!.yarn/cache
+
+.vscode