feat(server): added prometheus metrics to messaging server (#546)

Author: michaelmass

Dockerfile

@@ -36,5 +36,5 @@ RUN yarn workspaces focus --all --production
 
 ENV NODE_ENV=production
 
-ENTRYPOINT [ "yarn", "node", "./packages/server/dist/index.js" ]
-CMD []
+ENTRYPOINT [ "yarn", "node", "-r", "@bpinternal/trail/init", "./packages/server/dist/index.js" ]
+CMD []

packages/framework/src/entry.ts

@@ -20,6 +20,7 @@ export abstract class Entry {
   api: IApi
   stream: IStream
   socket: ISocket
+  interceptor: IInterceptor
 
   manager: ApiManager
   adminManager: AdminApiManager
@@ -29,13 +30,15 @@ export abstract class Entry {
     tapp: { new (): IApp },
     tapi: { new (app: IApp & any, manager: ApiManager, adminManager: AdminApiManager, express: Express): IApi },
     tstream: { new (app: IApp & any): IStream },
-    tsocket: { new (app: IApp & any): ISocket }
+    tsocket: { new (app: IApp & any): ISocket },
+    tinterceptor: { new (app: IApp & any, express: Express): IInterceptor }
   ) {
     this.router = express()
     this.router.disable('x-powered-by')
     this.routes = new Routes(this.router)
 
     this.app = new tapp()
+    this.interceptor = new tinterceptor(this.app, this.router)
 
     const auth = new Auth(this.app.clientTokens)
     this.manager = new ApiManager(this.routes.router, auth)
@@ -49,6 +52,8 @@ export abstract class Entry {
   }
 
   async setup() {
+    await this.interceptor.setup()
+
     await this.app.prepare(this.package, this.migrations)
     await this.app.setup()
     await this.app.postSetup()
@@ -98,3 +103,7 @@ interface ISocket {
   start(server: Server): Promise<void>
   destroy(): Promise<void>
 }
+
+interface IInterceptor {
+  setup(): Promise<void>
+}

packages/server/package.json

@@ -6,10 +6,11 @@
   "scripts": {
     "build": "yarn && yarn run -T tsc --build",
     "watch": "yarn && yarn run -T tsc --build --watch",
-    "start": "node ./dist/index.js",
+    "start": "node -r @bpinternal/trail/init ./dist/index.js",
     "dev": "yarn && yarn run -T ts-node-dev --debounce 500 --transpile-only ./src/index.ts"
   },
   "devDependencies": {
+    "@promster/types": "^3.2.4",
     "@types/cli-color": "^2.0.2",
     "@types/cors": "^2.8.12",
     "@types/express": "^4.17.13",
@@ -26,7 +27,8 @@
     "@botpress/messaging-channels-legacy": "npm:@botpress/[email protected]",
     "@botpress/messaging-engine": "1.2.0",
     "@botpress/messaging-framework": "1.2.0",
-    "@bpinternal/trail": "^0.0.1",
+    "@bpinternal/promex": "^0.5.0",
+    "@bpinternal/trail": "^0.1.0",
     "@opentelemetry/instrumentation-aws-sdk": "0.7.0",
     "@opentelemetry/instrumentation-express": "0.27.0",
     "@opentelemetry/instrumentation-http": "0.27.0",

packages/server/src/app.ts

@@ -8,6 +8,7 @@ import { HealthService } from './health/service'
 import { InstanceService } from './instances/service'
 import { MappingService } from './mapping/service'
 import { MessageService } from './messages/service'
+import { MetricsService } from './metrics/service'
 import { ProviderService } from './providers/service'
 import { ProvisionService } from './provisions/service'
 import { SocketService } from './socket/service'
@@ -35,6 +36,7 @@ export class App extends Framework {
   health: HealthService
   sockets: SocketService
   billing: BillingService
+  metrics: MetricsService
 
   constructor() {
     super()
@@ -93,6 +95,7 @@ export class App extends Framework {
     )
     this.sockets = new SocketService(this.caching, this.users)
     this.billing = new BillingService(this.logger, this.conversations, this.messages)
+    this.metrics = new MetricsService(this.conversations, this.messages)
   }
 
   async setup() {
@@ -112,6 +115,7 @@ export class App extends Framework {
     await this.instances.setup()
     await this.health.setup()
     await this.sockets.setup()
+    await this.metrics.setup()
   }
 
   async postSetup() {
@@ -130,5 +134,6 @@ export class App extends Framework {
     await super.destroy()
     await this.billing?.destroy()
     await this.instances?.destroy()
+    await this.metrics?.destroy()
   }
 }

packages/server/src/global.ts

@@ -10,6 +10,8 @@ export type MessagingEnv = FrameworkEnv & {
   ENABLE_BILLING_STATS?: string
   DISABLE_SOCKETS?: string
   ENABLE_LEGACY_CHANNELS?: string
+  METRICS_ENABLED?: string
+  METRICS_PORT?: string
 }
 
 declare global {

packages/server/src/index.ts

@@ -1,15 +1,11 @@
-/* eslint-disable import/order */
-import { init } from '@bpinternal/trail'
-init()
-
 import './rewire'
 import { Entry, start } from '@botpress/messaging-framework'
 import { Api } from './api'
 import { App } from './app'
+import { Interceptor } from './interceptor'
 import { Migrations } from './migrations'
 import { Socket } from './socket'
 import { Stream } from './stream'
-/* eslint-enable import/order */
 
 export class Root extends Entry {
   get name() {
@@ -29,7 +25,7 @@ export class Root extends Entry {
   }
 
   constructor() {
-    super(App, Api, Stream, Socket)
+    super(App, Api, Stream, Socket, Interceptor)
   }
 }
 

packages/server/src/interceptor.ts

@@ -0,0 +1,10 @@
+import type { Express } from 'express'
+import { App } from './app'
+
+export class Interceptor {
+  constructor(private app: App, private express: Express) {}
+
+  async setup() {
+    this.app.metrics.init(this.express)
+  }
+}

packages/server/src/metrics/service.ts

@@ -0,0 +1,67 @@
+import { Service } from '@botpress/messaging-engine'
+import * as promex from '@bpinternal/promex'
+import { defaultNormalizers } from '@promster/express'
+import type { Express } from 'express'
+import client from 'prom-client'
+import yn from 'yn'
+import { ConversationEvents } from '../conversations/events'
+import { ConversationService } from '../conversations/service'
+import { MessageEvents } from '../messages/events'
+import { MessageService } from '../messages/service'
+
+const messagesCount = new client.Counter({
+  name: 'messages_total',
+  help: 'Counter of all messages.'
+})
+
+const conversationsCount = new client.Counter({
+  name: 'conversations_total',
+  help: 'Counter of all conversations.'
+})
+
+export class MetricsService extends Service {
+  private metricsEnabled: boolean
+  private port: number
+
+  constructor(private conversations: ConversationService, private messages: MessageService) {
+    super()
+    this.metricsEnabled = yn(process.env.METRICS_ENABLED) ?? false
+    this.port = process.env.METRICS_PORT ? parseInt(process.env.METRICS_PORT) : 9090
+  }
+
+  async destroy() {
+    if (!this.metricsEnabled) {
+      return
+    }
+
+    await promex.stop()
+  }
+
+  init(app: Express) {
+    if (!this.metricsEnabled) {
+      return
+    }
+
+    promex.config({ normalizePath: defaultNormalizers.normalizePath })
+    promex.init(app)
+  }
+
+  async setup() {
+    if (!this.metricsEnabled) {
+      return
+    }
+
+    this.messages.events.on(MessageEvents.Created, this.handleMessageCreated.bind(this))
+    this.conversations.events.on(ConversationEvents.Created, this.handleConversationCreated.bind(this))
+
+    await promex.start({ port: this.port })
+  }
+
+  private async handleMessageCreated() {
+    messagesCount.inc()
+  }
+
+  private async handleConversationCreated() {
+    conversationsCount.inc()
+  }
+}