Performance monitoring

packages/core/core/src/aggregate/get-aggregates-interop-builder.ts

@@ -17,8 +17,6 @@ import {
   CommandResult,
 } from '../types/core'
 
-import { makeMonitoringSafe } from '../helpers'
-
 type AggregateData = {
   aggregateVersion: number
   aggregateId: string
@@ -372,13 +370,24 @@ const executeCommand = async (
   runtime: AggregateRuntime,
   command: Command
 ): Promise<CommandResult> => {
-  const monitoring = makeMonitoringSafe(runtime.monitoring)
+  const monitoringGroup =
+    runtime.monitoring != null
+      ? runtime.monitoring
+          .group({ Part: 'Command' })
+          .group({ AggregateName: command.aggregateName })
+          .group({ Type: command.type })
+      : null
+
+  if (monitoringGroup != null) {
+    monitoringGroup.time('Execution')
+  }
+
   const { jwt: actualJwt, jwtToken: deprecatedJwt } = command
 
   const jwt = actualJwt || deprecatedJwt
 
   const subSegment = getPerformanceTracerSubsegment(
-    monitoring,
+    runtime.monitoring,
     'executeCommand'
   )
 
@@ -419,7 +428,7 @@ const executeCommand = async (
       context: CommandContext
     ): Promise<CommandResult> => {
       const subSegment = getPerformanceTracerSubsegment(
-        monitoring,
+        runtime.monitoring,
         'processCommand'
       )
       try {
@@ -483,7 +492,10 @@ const executeCommand = async (
     }
 
     await (async (): Promise<void> => {
-      const subSegment = getPerformanceTracerSubsegment(monitoring, 'saveEvent')
+      const subSegment = getPerformanceTracerSubsegment(
+        runtime.monitoring,
+        'saveEvent'
+      )
 
       try {
         return await saveEvent(runtime, aggregate, command, processedEvent)
@@ -498,9 +510,16 @@ const executeCommand = async (
     return processedEvent
   } catch (error) {
     subSegment.addError(error)
-    await monitoring?.error?.(error, 'command', { command })
+
+    if (monitoringGroup != null) {
+      monitoringGroup.group({ AggregateId: command.aggregateId }).error(error)
+    }
     throw error
   } finally {
+    if (monitoringGroup != null) {
+      monitoringGroup.timeEnd('Execution')
+    }
+
     subSegment.close()
   }
 }

packages/core/core/src/aggregate/types.ts

@@ -44,7 +44,7 @@ export type AggregateRuntimeHooks = {
 }
 
 export type AggregateRuntime = {
-  monitoring: Monitoring
+  monitoring?: Monitoring
   secretsManager: SecretsManager
   eventstore: Eventstore
   hooks?: AggregateRuntimeHooks

packages/core/core/src/helpers.ts

@@ -7,11 +7,11 @@ export function firstOfType<T>(
   return vars.find((i) => selector(i)) as T
 }
 
-const createSafeHandler = <T extends (...args: any[]) => Promise<void>>(
-  fn: (...args: Parameters<T>) => Promise<void>
-) => async (...args: Parameters<T>): Promise<void> => {
+const createSafeHandler = <T extends (...args: any[]) => void>(
+  fn: (...args: Parameters<T>) => void
+) => (...args: Parameters<T>): void => {
   try {
-    await fn(...args)
+    fn(...args)
   } catch (e) {}
 }
 

packages/core/core/src/read-model/get-read-models-interop-builder.ts

@@ -11,12 +11,20 @@ import { getPerformanceTracerSubsegment } from '../utils'
 import { ReadModelMeta } from '../types/runtime'
 import getLog from '../get-log'
 
-const monitoredError = async (
+const monitoredError = (
   runtime: ReadModelRuntime,
   error: Error,
-  meta: any
+  readModelName: string,
+  resolverName: string
 ) => {
-  await runtime.monitoring?.error?.(error, 'readModelResolver', meta)
+  if (runtime.monitoring != null) {
+    const monitoringGroup = runtime.monitoring
+      .group({ Part: 'ReadModelResolver' })
+      .group({ ReadModel: readModelName })
+      .group({ Resolver: resolverName })
+
+    monitoringGroup.error(error)
+  }
   return error
 }
 
@@ -43,16 +51,14 @@ const getReadModelInterop = (
     const invoker = resolverInvokerMap[resolver]
     if (invoker == null) {
       log.error(`unable to find invoker for the resolver`)
-      throw await monitoredError(
+      throw monitoredError(
         runtime,
         createHttpError(
           HttpStatusCodes.UnprocessableEntity,
           `Resolver "${resolver}" does not exist`
         ),
-        {
-          readModelName: name,
-          resolverName: resolver,
-        }
+        name,
+        resolver
       )
     }
 
@@ -69,6 +75,18 @@ const getReadModelInterop = (
         }
       )
 
+      const monitoringGroup =
+        monitoring != null
+          ? monitoring
+              .group({ Part: 'ReadModelResolver' })
+              .group({ ReadModel: name })
+              .group({ Resolver: resolver })
+          : null
+
+      if (monitoringGroup != null) {
+        monitoringGroup.time('Execution')
+      }
+
       try {
         log.debug(`invoking the resolver`)
         const data = await invoker(connection, args, {
@@ -84,12 +102,16 @@ const getReadModelInterop = (
         if (subSegment != null) {
           subSegment.addError(error)
         }
-        await monitoring?.error?.(error, 'readModelResolver', {
-          readModelName: name,
-          resolverName: resolver,
-        })
+
+        if (monitoringGroup != null) {
+          monitoringGroup.error(error)
+        }
         throw error
       } finally {
+        if (monitoringGroup != null) {
+          monitoringGroup.timeEnd('Execution')
+        }
+
         if (subSegment != null) {
           subSegment.close()
         }
@@ -113,10 +135,14 @@ const getReadModelInterop = (
     try {
       return await handler()
     } catch (error) {
-      await monitoring?.error?.(error, 'readModelProjection', {
-        readModelName: readModel.name,
-        eventType,
-      })
+      if (monitoring != null) {
+        const monitoringGroup = monitoring
+          .group({ Part: 'ReadModelProjection' })
+          .group({ ReadModel: readModel.name })
+          .group({ EventType: eventType })
+
+        monitoringGroup.error(error)
+      }
       throw error
     }
   }

packages/core/core/src/read-model/types.ts

@@ -10,7 +10,7 @@ export type ReadModelRuntimeEventHandler = () => Promise<void>
 
 export type ReadModelRuntime = {
   secretsManager: SecretsManager
-  monitoring: Monitoring
+  monitoring?: Monitoring
 }
 
 export type ReadModelInterop = {

packages/core/core/src/saga/get-sagas-interop-builder.ts

@@ -55,10 +55,16 @@ const getInterop = (
     try {
       return await handler()
     } catch (error) {
-      await runtime.monitoring?.error?.(error, 'readModelProjection', {
-        readModelName: saga.name,
-        eventType,
-      })
+      if (runtime.monitoring != null) {
+        const monitoringGroup = runtime.monitoring
+          .group({
+            Part: 'SagaProjection',
+          })
+          .group({ Saga: saga.name })
+          .group({ EventType: eventType })
+
+        monitoringGroup.error(error)
+      }
       throw error
     }
   }

packages/core/core/src/types/runtime.ts

@@ -11,7 +11,6 @@ import {
   AggregateProjection,
   EventHandlerEncryptionFactory,
   ReadModelResolvers,
-  Command,
 } from './core'
 
 export type PerformanceSubsegment = {
@@ -27,6 +26,15 @@ export type PerformanceTracer = {
   getSegment: () => PerformanceSegment
 }
 
+export type Monitoring = {
+  group: (config: Record<string, any>) => Monitoring
+  error: (error: Error) => void
+  time: (name: string, timestamp?: number) => void
+  timeEnd: (name: string, timestamp?: number) => void
+  publish: () => Promise<void>
+  performance?: PerformanceTracer
+}
+
 export type Eventstore = {
   saveEvent: (event: any) => Promise<void>
   saveSnapshot: Function
@@ -82,37 +90,3 @@ export type ViewModelMeta = {
   encryption: EventHandlerEncryptionFactory
   invariantHash: string
 }
-
-export type MonitoringPartMap = {
-  command: {
-    command: Command
-  }
-  readModelProjection: {
-    readModelName: string
-    eventType: string
-  }
-  readModelResolver: {
-    readModelName: string
-    resolverName: string
-  }
-  viewModelProjection: {
-    name: string
-    eventType: string
-  }
-  viewModelResolver: {
-    name: string
-  }
-}
-export type MonitoringPart = keyof MonitoringPartMap
-export type MonitoringMeta<
-  TPart extends MonitoringPart
-> = MonitoringPartMap[TPart]
-
-export type Monitoring = {
-  error?: <K extends MonitoringPart, U extends MonitoringMeta<K>>(
-    error: Error,
-    part: K,
-    meta: U
-  ) => Promise<void>
-  performance?: PerformanceTracer
-}

packages/core/core/src/view-model/get-view-models-interop-builder.ts

@@ -111,10 +111,15 @@ const buildViewModel = async (
     } catch (error) {
       subSegment.addError(error)
       log.error(error.message)
-      await monitoring?.error?.(error, 'viewModelProjection', {
-        name,
-        eventType: event.type,
-      })
+
+      if (monitoring != null) {
+        const monitoringGroup = monitoring
+          .group({ Part: 'ViewModelProjection' })
+          .group({ ViewModel: name })
+          .group({ EventType: event.type })
+
+        monitoringGroup.error(error)
+      }
       throw error
     } finally {
       subSegment.close()
@@ -220,9 +225,13 @@ const getViewModelInterop = (
     } catch (error) {
       subSegment.addError(error)
 
-      await monitoring?.error?.(error, 'viewModelResolver', {
-        name,
-      })
+      if (monitoring != null) {
+        const monitoringGroup = monitoring
+          .group({ Part: 'ViewModelResolver' })
+          .group({ ViewModel: name })
+
+        monitoringGroup.error(error)
+      }
       throw error
     } finally {
       subSegment.close()

packages/core/core/src/view-model/types.ts

@@ -17,7 +17,7 @@ export type ViewModelBuildResult = {
 }
 
 export type ViewModelRuntime = {
-  monitoring: Monitoring
+  monitoring?: Monitoring
   eventstore: Eventstore
   secretsManager: SecretsManager
 }