✨ [RUMF-1071] forward internal monitoring to bridge

packages/core/src/domain/internalMonitoring/index.ts

@@ -0,0 +1,14 @@
+export {
+  InternalMonitoring,
+  MonitoringMessage,
+  monitored,
+  monitor,
+  callMonitored,
+  addMonitoringMessage,
+  addMonitoringError,
+  startFakeInternalMonitoring,
+  resetInternalMonitoring,
+  setDebugMode,
+  startInternalMonitoring,
+} from './internalMonitoring'
+export { startMonitoringBatch } from './startMonitoringBatch'

packages/core/src/domain/internalMonitoring.spec.ts → packages/core/src/domain/internalMonitoring/internalMonitoring.spec.ts

@@ -1,7 +1,14 @@
 import sinon from 'sinon'
-import { Clock, mockClock, stubEndpointBuilder } from '../../test/specHelper'
-
-import { Configuration } from './configuration'
+import { resetExperimentalFeatures, updateExperimentalFeatures } from '../..'
+import {
+  Clock,
+  deleteEventBridgeStub,
+  initEventBridgeStub,
+  mockClock,
+  stubEndpointBuilder,
+} from '../../../test/specHelper'
+
+import { Configuration } from '../configuration'
 import {
   InternalMonitoring,
   monitor,
@@ -171,13 +178,12 @@ describe('internal monitoring', () => {
     })
   })
 
-  describe('request', () => {
+  describe('transport', () => {
     const FAKE_DATE = 123456
     let server: sinon.SinonFakeServer
     let clock: Clock
 
     beforeEach(() => {
-      startInternalMonitoring(configuration as Configuration)
       server = sinon.fakeServer.create()
       clock = mockClock(new Date(FAKE_DATE))
     })
@@ -186,9 +192,13 @@ describe('internal monitoring', () => {
       resetInternalMonitoring()
       server.restore()
       clock.cleanup()
+      resetExperimentalFeatures()
+      deleteEventBridgeStub()
     })
 
     it('should send the needed data', () => {
+      startInternalMonitoring(configuration as Configuration)
+
       callMonitored(() => {
         throw new Error('message')
       })
@@ -205,6 +215,8 @@ describe('internal monitoring', () => {
     })
 
     it('should cap the data sent', () => {
+      startInternalMonitoring(configuration as Configuration)
+
       const max = configuration.maxInternalMonitoringMessagesPerPage!
       for (let i = 0; i < max + 3; i += 1) {
         callMonitored(() => {
@@ -214,6 +226,19 @@ describe('internal monitoring', () => {
 
       expect(server.requests.length).toEqual(max)
     })
+
+    it('should send bridge event when bridge is present', () => {
+      updateExperimentalFeatures(['event-bridge'])
+      const sendSpy = spyOn(initEventBridgeStub(), 'send')
+      startInternalMonitoring(configuration as Configuration)
+
+      callMonitored(() => {
+        throw new Error('message')
+      })
+
+      expect(server.requests.length).toEqual(0)
+      expect(sendSpy).toHaveBeenCalled()
+    })
   })
 
   describe('external context', () => {

packages/core/src/domain/internalMonitoring.ts → packages/core/src/domain/internalMonitoring/internalMonitoring.ts

@@ -1,11 +1,11 @@
-import { Context } from '../tools/context'
-import { display } from '../tools/display'
-import { toStackTraceString } from '../tools/error'
-import { assign, combine, jsonStringify, Parameters, ThisParameterType } from '../tools/utils'
-import { Batch, HttpRequest } from '../transport'
-import { Configuration } from './configuration'
-import { EndpointBuilder } from './configuration/endpointBuilder'
-import { computeStackTrace } from './tracekit'
+import { Context } from '../../tools/context'
+import { display } from '../../tools/display'
+import { toStackTraceString } from '../../tools/error'
+import { assign, combine, jsonStringify, Parameters, ThisParameterType } from '../../tools/utils'
+import { canUseEventBridge, getEventBridge } from '../../transport'
+import { Configuration } from '../configuration'
+import { computeStackTrace } from '../tracekit'
+import { startMonitoringBatch } from '../internalMonitoring/startMonitoringBatch'
 
 enum StatusType {
   info = 'info',
@@ -26,85 +26,61 @@ export interface MonitoringMessage extends Context {
 }
 
 const monitoringConfiguration: {
-  batch?: Batch
   debugMode?: boolean
   maxMessagesPerPage: number
   sentMessageCount: number
 } = { maxMessagesPerPage: 0, sentMessageCount: 0 }
 
-let externalContextProvider: () => Context
+let onInternalMonitoringEventCollected: ((message: MonitoringMessage) => void) | undefined
 
 export function startInternalMonitoring(configuration: Configuration): InternalMonitoring {
-  if (configuration.internalMonitoringEndpointBuilder) {
-    const batch = startMonitoringBatch(configuration)
-
-    assign(monitoringConfiguration, {
-      batch,
-      maxMessagesPerPage: configuration.maxInternalMonitoringMessagesPerPage,
-      sentMessageCount: 0,
-    })
-  }
-  return {
-    setExternalContextProvider: (provider: () => Context) => {
-      externalContextProvider = provider
-    },
-  }
-}
+  let externalContextProvider: () => Context
 
-function startMonitoringBatch(configuration: Configuration) {
-  const primaryBatch = createMonitoringBatch(configuration.internalMonitoringEndpointBuilder!)
-  let replicaBatch: Batch | undefined
-  if (configuration.replica !== undefined) {
-    replicaBatch = createMonitoringBatch(configuration.replica.internalMonitoringEndpointBuilder)
+  if (canUseEventBridge()) {
+    const bridge = getEventBridge()!
+    onInternalMonitoringEventCollected = (message: MonitoringMessage) =>
+      bridge.send('internal_log', withContext(message))
+  } else if (configuration.internalMonitoringEndpointBuilder) {
+    const batch = startMonitoringBatch(configuration)
+    onInternalMonitoringEventCollected = (message: MonitoringMessage) => batch.add(withContext(message))
   }
 
-  function createMonitoringBatch(endpointBuilder: EndpointBuilder) {
-    return new Batch(
-      new HttpRequest(endpointBuilder, configuration.batchBytesLimit),
-      configuration.maxBatchSize,
-      configuration.batchBytesLimit,
-      configuration.maxMessageSize,
-      configuration.flushTimeout
-    )
-  }
+  assign(monitoringConfiguration, {
+    maxMessagesPerPage: configuration.maxInternalMonitoringMessagesPerPage,
+    sentMessageCount: 0,
+  })
 
   function withContext(message: MonitoringMessage) {
     return combine(
-      {
-        date: new Date().getTime(),
-      },
+      { date: new Date().getTime() },
       externalContextProvider !== undefined ? externalContextProvider() : {},
       message
     )
   }
 
   return {
-    add(message: MonitoringMessage) {
-      const contextualizedMessage = withContext(message)
-      primaryBatch.add(contextualizedMessage)
-      if (replicaBatch) {
-        replicaBatch.add(contextualizedMessage)
-      }
+    setExternalContextProvider: (provider: () => Context) => {
+      externalContextProvider = provider
     },
   }
 }
 
 export function startFakeInternalMonitoring() {
   const messages: MonitoringMessage[] = []
   assign(monitoringConfiguration, {
-    batch: {
-      add(message: MonitoringMessage) {
-        messages.push(message)
-      },
-    },
     maxMessagesPerPage: Infinity,
     sentMessageCount: 0,
   })
+
+  onInternalMonitoringEventCollected = (message: MonitoringMessage) => {
+    messages.push(message)
+  }
+
   return messages
 }
 
 export function resetInternalMonitoring() {
-  monitoringConfiguration.batch = undefined
+  onInternalMonitoringEventCollected = undefined
 }
 
 export function monitored<T extends (...params: any[]) => unknown>(
@@ -114,7 +90,7 @@ export function monitored<T extends (...params: any[]) => unknown>(
 ) {
   const originalMethod = descriptor.value!
   descriptor.value = function (this: any, ...args: Parameters<T>) {
-    const decorated = monitoringConfiguration.batch ? monitor(originalMethod) : originalMethod
+    const decorated = onInternalMonitoringEventCollected ? monitor(originalMethod) : originalMethod
     return decorated.apply(this, args) as ReturnType<T>
   } as T
 }
@@ -143,7 +119,7 @@ export function callMonitored<T extends (...args: any[]) => any>(
   } catch (e) {
     logErrorIfDebug(e)
     try {
-      addErrorToMonitoringBatch(e)
+      addMonitoringError(e)
     } catch (e) {
       logErrorIfDebug(e)
     }
@@ -152,28 +128,27 @@ export function callMonitored<T extends (...args: any[]) => any>(
 
 export function addMonitoringMessage(message: string, context?: Context) {
   logMessageIfDebug(message, context)
-  addToMonitoringBatch({
+  addToMonitoring({
     message,
     ...context,
     status: StatusType.info,
   })
 }
 
-export function addErrorToMonitoringBatch(e: unknown) {
-  addToMonitoringBatch({
+export function addMonitoringError(e: unknown) {
+  addToMonitoring({
     ...formatError(e),
     status: StatusType.error,
   })
 }
 
-function addToMonitoringBatch(message: MonitoringMessage) {
+function addToMonitoring(message: MonitoringMessage) {
   if (
-    monitoringConfiguration.batch &&
+    onInternalMonitoringEventCollected &&
     monitoringConfiguration.sentMessageCount < monitoringConfiguration.maxMessagesPerPage
   ) {
     monitoringConfiguration.sentMessageCount += 1
-
-    monitoringConfiguration.batch.add(message)
+    onInternalMonitoringEventCollected(message)
   }
 }
 

packages/core/src/domain/internalMonitoring/startMonitoringBatch.ts

@@ -0,0 +1,29 @@
+import { Batch, HttpRequest } from '../../transport'
+import { Configuration, EndpointBuilder, MonitoringMessage } from '../..'
+
+export function startMonitoringBatch(configuration: Configuration) {
+  const primaryBatch = createMonitoringBatch(configuration.internalMonitoringEndpointBuilder!)
+  let replicaBatch: Batch | undefined
+  if (configuration.replica !== undefined) {
+    replicaBatch = createMonitoringBatch(configuration.replica.internalMonitoringEndpointBuilder)
+  }
+
+  function createMonitoringBatch(endpointBuilder: EndpointBuilder) {
+    return new Batch(
+      new HttpRequest(endpointBuilder, configuration.batchBytesLimit),
+      configuration.maxBatchSize,
+      configuration.batchBytesLimit,
+      configuration.maxMessageSize,
+      configuration.flushTimeout
+    )
+  }
+
+  return {
+    add(message: MonitoringMessage) {
+      primaryBatch.add(message)
+      if (replicaBatch) {
+        replicaBatch.add(message)
+      }
+    },
+  }
+}

packages/core/src/index.ts

@@ -22,7 +22,7 @@ export {
   monitor,
   callMonitored,
   addMonitoringMessage,
-  addErrorToMonitoringBatch,
+  addMonitoringError,
   startFakeInternalMonitoring,
   resetInternalMonitoring,
   setDebugMode,

packages/core/src/transport/eventBridge.ts

@@ -28,6 +28,7 @@ export function getEventBridge<T, E>() {
 
 export function canUseEventBridge(): boolean {
   const bridge = getEventBridge()
+
   return !!bridge && includes(bridge.getAllowedWebViewHosts(), window.location.hostname)
 }
 

packages/core/src/transport/httpRequest.ts

@@ -1,5 +1,5 @@
 import { EndpointBuilder } from '../domain/configuration/endpointBuilder'
-import { monitor, addErrorToMonitoringBatch, addMonitoringMessage } from '../domain/internalMonitoring'
+import { monitor, addMonitoringError, addMonitoringMessage } from '../domain/internalMonitoring'
 
 let hasReportedXhrError = false
 
@@ -69,6 +69,6 @@ let hasReportedBeaconError = false
 function reportBeaconError(e: unknown) {
   if (!hasReportedBeaconError) {
     hasReportedBeaconError = true
-    addErrorToMonitoringBatch(e)
+    addMonitoringError(e)
   }
 }

packages/rum/src/domain/segmentCollection/startDeflateWorker.ts

@@ -1,4 +1,4 @@
-import { addErrorToMonitoringBatch, display, includes, monitor } from '@datadog/browser-core'
+import { addMonitoringError, display, includes, monitor } from '@datadog/browser-core'
 import { createDeflateWorker, DeflateWorker } from './deflateWorker'
 
 /**
@@ -102,11 +102,11 @@ function onError(error: ErrorEvent | Error | string) {
           'https://docs.datadoghq.com/real_user_monitoring/faq/content_security_policy'
       )
     } else {
-      addErrorToMonitoringBatch(error)
+      addMonitoringError(error)
     }
     state.callbacks.forEach((callback) => callback())
     state = { status: DeflateWorkerStatus.Error }
   } else {
-    addErrorToMonitoringBatch(error)
+    addMonitoringError(error)
   }
 }

test/e2e/lib/framework/pageSetups.ts

@@ -166,9 +166,10 @@ function setupEventBridge(servers: Servers) {
           return '["${baseHostname}"]'
         },
         send(e) {
-          const { event } = JSON.parse(e)
+          const { eventType, event } = JSON.parse(e)
           const request = new XMLHttpRequest()
-          request.open('POST', '${servers.intake.url}/v1/input/rum?bridge=1', true)
+          const endpoint = eventType === 'internal_log' ? 'internalMonitoring' : 'rum'
+          request.open('POST', \`${servers.intake.url}/v1/input/\${endpoint}?bridge=1\`, true)
           request.send(JSON.stringify(event))
         },
       }

test/e2e/scenario/rum/eventBridge.scenario.ts

@@ -60,4 +60,23 @@ describe('bridge present', () => {
       expect(serverEvents.rumViews.length).toEqual(0)
       expect(bridgeEvents.rumViews.length).toBeGreaterThan(0)
     })
+
+  createTest('forward internal monitoring to the bridge')
+    .withLogs({ enableExperimentalFeatures: ['event-bridge'], internalMonitoringApiKey: 'xxx' })
+    .withEventBridge()
+    .run(async ({ serverEvents, bridgeEvents }) => {
+      await browserExecute(() => {
+        const context = {
+          get foo() {
+            throw new window.Error('bar')
+          },
+        }
+        // eslint-disable-next-line @typescript-eslint/no-unnecessary-type-assertion
+        window.DD_LOGS!.logger.log('hop', context as any)
+      })
+
+      await flushEvents()
+      expect(serverEvents.internalMonitoring.length).toBe(0)
+      expect(bridgeEvents.internalMonitoring.length).toBe(1)
+    })
 })