✨ enable rum profiler compression when possible (#3861)

Author: thomas-lebeau

packages/core/package.json

@@ -18,6 +18,10 @@
     "url": "https://github.com/DataDog/browser-sdk.git",
     "directory": "packages/core"
   },
+  "devDependencies": {
+    "@types/pako": "2.0.4",
+    "pako": "2.1.0"
+  },
   "volta": {
     "extends": "../../package.json"
   },

packages/core/src/domain/deflate/deflate.types.ts

@@ -53,4 +53,5 @@ export const enum DeflateEncoderStreamId {
   REPLAY = 1,
   RUM = 2,
   TELEMETRY = 4,
+  PROFILING = 6,
 }

packages/core/test/index.ts

@@ -29,3 +29,4 @@ export * from './wait'
 export * from './consoleLog'
 export * from './createHooks'
 export * from './fakeSessionStoreStrategy'
+export * from './readFormData'

packages/core/test/interceptRequests.ts

@@ -1,6 +1,7 @@
 import type { EndpointBuilder } from '../src'
 import { INTAKE_URL_PARAMETERS, noop } from '../src'
 import { mockXhr, MockXhr } from './emulate/mockXhr'
+import { readFormData } from './readFormData'
 import { registerCleanupTask } from './registerCleanupTask'
 
 const INTAKE_PARAMS = INTAKE_URL_PARAMETERS.join('&')
@@ -107,3 +108,6 @@ export function interceptRequests() {
     },
   }
 }
+export function readFormDataRequest<T>(request: Request): Promise<T> {
+  return readFormData(request.body as unknown as FormData)
+}

packages/core/test/readFormData.ts

@@ -0,0 +1,51 @@
+import { inflate } from 'pako'
+
+// Zlib streams using a default compression are starting with bytes 120 156 (0x78 0x9c)
+// https://stackoverflow.com/a/9050274
+const Z_LIB_MAGIC_BYTES = 0x789c
+
+export async function readFormData<T>(formData: FormData): Promise<T> {
+  const entries = getEntries(formData)
+  const data: Record<string, unknown> = {}
+
+  for (const [key, value] of Object.entries(entries)) {
+    if (value instanceof Blob) {
+      data[key] = await readJsonBlob(value)
+    } else {
+      data[key] = value
+    }
+  }
+
+  return data as T
+}
+
+function getEntries(payload: FormData) {
+  const entries = {} as Record<string, FormDataEntryValue>
+  payload.forEach((data, key) => {
+    entries[key] = data
+  })
+  return entries
+}
+
+function isZlibCompressed(buffer: ArrayBuffer) {
+  return new DataView(buffer).getUint16(0) === Z_LIB_MAGIC_BYTES
+}
+
+function readJsonBlob<T>(blob: Blob): Promise<T> {
+  // Safari Mobile 14 should support blob.text() or blob.arrayBuffer() but the APIs are not defined on the safari
+  // provided by browserstack, so we still need to use a FileReader for now.
+  // https://caniuse.com/mdn-api_blob_arraybuffer
+  // https://caniuse.com/mdn-api_blob_text
+  return new Promise((resolve) => {
+    const reader = new FileReader()
+    reader.addEventListener('loadend', () => {
+      const buffer = reader.result as ArrayBuffer
+      const decompressed = isZlibCompressed(buffer) ? inflate(buffer) : buffer
+      const decoded = new TextDecoder().decode(decompressed)
+      const deserialized = JSON.parse(decoded)
+
+      resolve(deserialized as T)
+    })
+    reader.readAsArrayBuffer(blob)
+  })
+}

packages/rum-core/src/boot/rumPublicApi.ts

@@ -13,6 +13,7 @@ import type {
   Account,
   RumInternalContext,
   Telemetry,
+  Encoder,
 } from '@datadog/browser-core'
 import {
   ContextManagerMethod,
@@ -484,7 +485,8 @@ export interface ProfilerApi {
     hooks: Hooks,
     configuration: RumConfiguration,
     sessionManager: RumSessionManager,
-    viewHistory: ViewHistory
+    viewHistory: ViewHistory,
+    createEncoder: (streamId: DeflateEncoderStreamId) => Encoder
   ) => void
 }
 
@@ -544,14 +546,17 @@ export function makeRumPublicApi(
     trackingConsentState,
     customVitalsState,
     (configuration, deflateWorker, initialViewOptions) => {
+      const createEncoder =
+        deflateWorker && options.createDeflateEncoder
+          ? (streamId: DeflateEncoderStreamId) => options.createDeflateEncoder!(configuration, deflateWorker, streamId)
+          : createIdentityEncoder
+
       const startRumResult = startRumImpl(
         configuration,
         recorderApi,
         profilerApi,
         initialViewOptions,
-        deflateWorker && options.createDeflateEncoder
-          ? (streamId) => options.createDeflateEncoder!(configuration, deflateWorker, streamId)
-          : createIdentityEncoder,
+        createEncoder,
         trackingConsentState,
         customVitalsState,
         bufferedDataObservable,
@@ -572,7 +577,8 @@ export function makeRumPublicApi(
         startRumResult.hooks,
         configuration,
         startRumResult.session,
-        startRumResult.viewHistory
+        startRumResult.viewHistory,
+        createEncoder
       )
 
       strategy = createPostStartStrategy(strategy, startRumResult)

packages/rum-core/src/index.ts

@@ -71,3 +71,5 @@ export type { Hooks, DefaultRumEventAttributes, DefaultTelemetryEventAttributes
 export { createHooks } from './domain/hooks'
 export { isSampled } from './domain/sampler/sampler'
 export type { TracingOption, PropagatorType } from './domain/tracing/tracer.types'
+export type { TransportPayload } from './transport/formDataTransport'
+export { createFormDataTransport } from './transport/formDataTransport'

packages/rum-core/src/transport/formDataTransport.spec.ts

@@ -0,0 +1,35 @@
+import { createIdentityEncoder, DeflateEncoderStreamId as CoreDeflateEncoderStreamId } from '@datadog/browser-core'
+import { interceptRequests, readFormDataRequest } from '@datadog/browser-core/test'
+import { LifeCycle } from '../domain/lifeCycle'
+import { mockRumConfiguration } from '../../test'
+import { createFormDataTransport } from './formDataTransport'
+
+describe('createFormDataTransport', () => {
+  function setup() {
+    const interceptor = interceptRequests()
+    const lifeCycle = new LifeCycle()
+    const transport = createFormDataTransport(
+      mockRumConfiguration(),
+      lifeCycle,
+      createIdentityEncoder,
+      CoreDeflateEncoderStreamId.REPLAY
+    )
+
+    return { interceptor, transport, lifeCycle }
+  }
+
+  it('should send event and attachments as FormData', async () => {
+    const { interceptor, transport } = setup()
+
+    const payload = {
+      event: { type: 'test-event', data: 'test-data' },
+      'attachment.json': { foo: 'bar' },
+    }
+
+    await transport.send(payload)
+
+    expect(interceptor.requests).toHaveSize(1)
+    expect(interceptor.requests[0].body).toBeInstanceOf(FormData)
+    expect(await readFormDataRequest(interceptor.requests[0])).toEqual(payload)
+  })
+})

packages/rum-core/src/transport/formDataTransport.ts

@@ -0,0 +1,89 @@
+import type {
+  Uint8ArrayBuffer,
+  Encoder,
+  EncoderResult,
+  DeflateEncoderStreamId,
+  RawError,
+  Context,
+} from '@datadog/browser-core'
+import { addTelemetryDebug, createHttpRequest, jsonStringify, objectEntries } from '@datadog/browser-core'
+import type { RumConfiguration } from '../domain/configuration'
+import type { LifeCycle } from '../domain/lifeCycle'
+import { LifeCycleEventType } from '../domain/lifeCycle'
+
+/**
+ * transport payload consist of an event and one or more attachments
+ */
+export interface TransportPayload {
+  event: Context
+  [key: string]: Context
+}
+
+export interface Transport<T extends TransportPayload> {
+  send: (data: T) => Promise<void>
+}
+
+export function createFormDataTransport<T extends TransportPayload>(
+  configuration: RumConfiguration,
+  lifeCycle: LifeCycle,
+  createEncoder: (streamId: DeflateEncoderStreamId) => Encoder,
+  streamId: DeflateEncoderStreamId
+) {
+  const reportError = (error: RawError) => {
+    lifeCycle.notify(LifeCycleEventType.RAW_ERROR_COLLECTED, { error })
+
+    // monitor-until: forever, to keep an eye on the errors reported to customers
+    addTelemetryDebug('Error reported to customer', { 'error.message': error.message })
+  }
+
+  const httpRequest = createHttpRequest(
+    [configuration.profilingEndpointBuilder],
+    configuration.batchBytesLimit,
+    reportError
+  )
+
+  const encoder = createEncoder(streamId)
+
+  return {
+    async send({ event, ...attachments }: T) {
+      const formData = new FormData()
+      const serializedEvent = jsonStringify(event)
+
+      if (!serializedEvent) {
+        throw new Error('Failed to serialize event')
+      }
+
+      formData.append('event', new Blob([serializedEvent], { type: 'application/json' }), 'event.json')
+
+      let bytesCount = serializedEvent.length
+
+      for (const [key, value] of objectEntries(attachments as Record<string, Context>)) {
+        const serializedValue = jsonStringify(value)
+
+        if (!serializedValue) {
+          throw new Error('Failed to serialize attachment')
+        }
+
+        const result = await encode(encoder, serializedValue)
+
+        bytesCount += result.outputBytesCount
+        formData.append(key, new Blob([result.output]), key)
+      }
+
+      httpRequest.send({
+        data: formData,
+        bytesCount,
+      })
+    },
+  }
+}
+
+function encode<T extends string | Uint8ArrayBuffer>(encoder: Encoder<T>, data: string): Promise<EncoderResult<T>> {
+  return new Promise((resolve) => {
+    encoder.write(data)
+
+    encoder.finish((encoderResult) => {
+      resolve(encoderResult)
+    })
+  })
+}

packages/rum/src/boot/profilerApi.ts

@@ -6,6 +6,7 @@ import type {
   ProfilerApi,
   Hooks,
 } from '@datadog/browser-rum-core'
+import type { DeflateEncoderStreamId, Encoder } from '@datadog/browser-core'
 import { isSampled } from '@datadog/browser-rum-core'
 import { addTelemetryDebug, monitorError } from '@datadog/browser-core'
 import type { RUMProfiler } from '../domain/profiling/types'
@@ -21,7 +22,8 @@ export function makeProfilerApi(): ProfilerApi {
     hooks: Hooks,
     configuration: RumConfiguration,
     sessionManager: RumSessionManager,
-    viewHistory: ViewHistory
+    viewHistory: ViewHistory,
+    createEncoder: (streamId: DeflateEncoderStreamId) => Encoder
   ) {
     const session = sessionManager.findTrackedSession() // Check if the session is tracked.
 
@@ -59,7 +61,7 @@ export function makeProfilerApi(): ProfilerApi {
           return
         }
 
-        profiler = createRumProfiler(configuration, lifeCycle, sessionManager, profilingContextManager)
+        profiler = createRumProfiler(configuration, lifeCycle, sessionManager, profilingContextManager, createEncoder)
         profiler.start(viewHistory.findView())
       })
       .catch(monitorError)