⚡♻️ simplify replica (#3552)

* ✨ httpRequest can send requests to multiple endpoints

* ♻️ use createBatch in telemetry

* ♻️ use createBatch in rum

* ♻️ use createBatch in logs

* 🔥 remove now unused startBatchWithReplica

Author: BenoitZugmeyer

packages/core/src/domain/configuration/endpointBuilder.spec.ts

@@ -46,6 +46,13 @@ describe('endpointBuilder', () => {
       expect(url).not.toContain('/rum?ddsource')
       expect(url).toContain('ddsource=browser')
     })
+
+    it('accepts extra parameters', () => {
+      const extraParameters = ['application.id=1234', 'application.version=1.0.0']
+      const url = createEndpointBuilder(initConfiguration, 'rum', extraParameters).build('fetch', DEFAULT_PAYLOAD)
+      expect(url).toContain('application.id=1234')
+      expect(url).toContain('application.version=1.0.0')
+    })
   })
 
   describe('proxy configuration', () => {

packages/core/src/domain/configuration/endpointBuilder.ts

@@ -19,12 +19,16 @@ export type ApiType =
 
 export type EndpointBuilder = ReturnType<typeof createEndpointBuilder>
 
-export function createEndpointBuilder(initConfiguration: InitConfiguration, trackType: TrackType) {
+export function createEndpointBuilder(
+  initConfiguration: InitConfiguration,
+  trackType: TrackType,
+  extraParameters?: string[]
+) {
   const buildUrlWithParameters = createEndpointUrlWithParametersBuilder(initConfiguration, trackType)
 
   return {
     build(api: ApiType, payload: Payload) {
-      const parameters = buildEndpointParameters(initConfiguration, trackType, api, payload)
+      const parameters = buildEndpointParameters(initConfiguration, trackType, api, payload, extraParameters)
       return buildUrlWithParameters(parameters)
     },
     trackType,
@@ -84,15 +88,16 @@ function buildEndpointParameters(
   { clientToken, internalAnalyticsSubdomain }: InitConfiguration,
   trackType: TrackType,
   api: ApiType,
-  { retry, encoding }: Payload
+  { retry, encoding }: Payload,
+  extraParameters: string[] = []
 ) {
   const parameters = [
     'ddsource=browser',
     `dd-api-key=${clientToken}`,
     `dd-evp-origin-version=${encodeURIComponent(__BUILD_ENV__SDK_VERSION__)}`,
     'dd-evp-origin=browser',
     `dd-request-id=${generateUUID()}`,
-  ]
+  ].concat(extraParameters)
 
   if (encoding) {
     parameters.push(`dd-evp-encoding=${encoding}`)

packages/core/src/domain/configuration/transportConfiguration.spec.ts

@@ -50,6 +50,20 @@ describe('transportConfiguration', () => {
     })
   })
 
+  it('adds the replica application id to the rum replica endpoint', () => {
+    const replicaApplicationId = 'replica-application-id'
+    const configuration = computeTransportConfiguration({
+      clientToken,
+      replica: {
+        clientToken: 'replica-client-token',
+        applicationId: replicaApplicationId,
+      },
+    })
+    expect(configuration.replica!.rumEndpointBuilder.build('fetch', DEFAULT_PAYLOAD)).toContain(
+      `application.id=${replicaApplicationId}`
+    )
+  })
+
   describe('isIntakeUrl', () => {
     const v1IntakePath = `/v1/input/${clientToken}`
     ;[

packages/core/src/domain/configuration/transportConfiguration.ts

@@ -16,7 +16,6 @@ export interface TransportConfiguration {
 }
 
 export interface ReplicaConfiguration {
-  applicationId?: string
   logsEndpointBuilder: EndpointBuilder
   rumEndpointBuilder: EndpointBuilder
 }
@@ -55,12 +54,12 @@ function computeReplicaConfiguration(initConfiguration: InitConfiguration): Repl
     clientToken: initConfiguration.replica.clientToken,
   }
 
-  const replicaEndpointBuilders = {
+  return {
     logsEndpointBuilder: createEndpointBuilder(replicaConfiguration, 'logs'),
-    rumEndpointBuilder: createEndpointBuilder(replicaConfiguration, 'rum'),
+    rumEndpointBuilder: createEndpointBuilder(replicaConfiguration, 'rum', [
+      `application.id=${initConfiguration.replica.applicationId}`,
+    ]),
   }
-
-  return { applicationId: initConfiguration.replica.applicationId, ...replicaEndpointBuilders }
 }
 
 export function isIntakeUrl(url: string): boolean {

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

@@ -52,7 +52,5 @@ export type DeflateEncoder = Encoder<Uint8ArrayBuffer> & { stop: () => void }
 export const enum DeflateEncoderStreamId {
   REPLAY = 1,
   RUM = 2,
-  RUM_REPLICA = 3,
   TELEMETRY = 4,
-  TELEMETRY_REPLICA = 5,
 }

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

@@ -5,7 +5,6 @@ export {
   addTelemetryError,
   resetTelemetry,
   startTelemetry,
-  isTelemetryReplicationAllowed,
   addTelemetryConfiguration,
   addTelemetryUsage,
   addTelemetryMetrics,

packages/core/src/domain/telemetry/telemetry.ts

@@ -17,7 +17,13 @@ import { NonErrorPrefix } from '../error/error.types'
 import type { StackTrace } from '../../tools/stackTrace/computeStackTrace'
 import { computeStackTrace } from '../../tools/stackTrace/computeStackTrace'
 import { getConnectivity } from '../connectivity'
-import { canUseEventBridge, getEventBridge, startBatchWithReplica } from '../../transport'
+import {
+  canUseEventBridge,
+  createFlushController,
+  createHttpRequest,
+  getEventBridge,
+  createBatch,
+} from '../../transport'
 import type { Encoder } from '../../tools/encoder'
 import type { PageMayExitEvent } from '../../browser/pageMayExitObservable'
 import { DeflateEncoderStreamId } from '../deflate'
@@ -190,30 +196,30 @@ function startTelemetryTransport(
   if (canUseEventBridge()) {
     const bridge = getEventBridge<'internal_telemetry', TelemetryEvent>()!
     const telemetrySubscription = telemetryObservable.subscribe((event) => bridge.send('internal_telemetry', event))
-    cleanupTasks.push(() => telemetrySubscription.unsubscribe())
+    cleanupTasks.push(telemetrySubscription.unsubscribe)
   } else {
-    const telemetryBatch = startBatchWithReplica(
-      configuration,
-      {
-        endpoint: configuration.rumEndpointBuilder,
-        encoder: createEncoder(DeflateEncoderStreamId.TELEMETRY),
-      },
-      configuration.replica && {
-        endpoint: configuration.replica.rumEndpointBuilder,
-        encoder: createEncoder(DeflateEncoderStreamId.TELEMETRY_REPLICA),
-      },
-      reportError,
-      pageMayExitObservable,
-
-      // We don't use an actual session expire observable here, to make telemetry collection
-      // independent of the session. This allows to start and send telemetry events ealier.
-      new Observable()
-    )
-    cleanupTasks.push(() => telemetryBatch.stop())
-    const telemetrySubscription = telemetryObservable.subscribe((event) => {
-      telemetryBatch.add(event, isTelemetryReplicationAllowed(configuration))
+    const endpoints = [configuration.rumEndpointBuilder]
+    if (configuration.replica && isTelemetryReplicationAllowed(configuration)) {
+      endpoints.push(configuration.replica.rumEndpointBuilder)
+    }
+    const telemetryBatch = createBatch({
+      encoder: createEncoder(DeflateEncoderStreamId.TELEMETRY),
+      request: createHttpRequest(endpoints, configuration.batchBytesLimit, reportError),
+      flushController: createFlushController({
+        messagesLimit: configuration.batchMessagesLimit,
+        bytesLimit: configuration.batchBytesLimit,
+        durationLimit: configuration.flushTimeout,
+        pageMayExitObservable,
+
+        // We don't use an actual session expire observable here, to make telemetry collection
+        // independent of the session. This allows to start and send telemetry events earlier.
+        sessionExpireObservable: new Observable(),
+      }),
+      messageBytesLimit: configuration.messageBytesLimit,
     })
-    cleanupTasks.push(() => telemetrySubscription.unsubscribe())
+    cleanupTasks.push(telemetryBatch.stop)
+    const telemetrySubscription = telemetryObservable.subscribe(telemetryBatch.add)
+    cleanupTasks.push(telemetrySubscription.unsubscribe)
   }
 
   return {
@@ -236,7 +242,7 @@ export function resetTelemetry() {
  * Avoid mixing telemetry events from different data centers
  * but keep replicating staging events for reliability
  */
-export function isTelemetryReplicationAllowed(configuration: Configuration) {
+function isTelemetryReplicationAllowed(configuration: Configuration) {
   return configuration.site === INTAKE_SITE_STAGING
 }
 

packages/core/src/index.ts

@@ -47,7 +47,6 @@ export {
   addTelemetryError,
   resetTelemetry,
   TelemetryService,
-  isTelemetryReplicationAllowed,
   addTelemetryConfiguration,
   addTelemetryUsage,
   addTelemetryMetrics,
@@ -69,7 +68,7 @@ export {
   getEventBridge,
   bridgeSupports,
   BridgeCapability,
-  startBatchWithReplica,
+  createBatch,
   createFlushController,
 } from './transport'
 export * from './tools/display'

packages/core/src/transport/httpRequest.spec.ts

@@ -26,7 +26,7 @@ describe('httpRequest', () => {
     interceptor = interceptRequests()
     requests = interceptor.requests
     endpointBuilder = mockEndpointBuilder(ENDPOINT_URL)
-    request = createHttpRequest(endpointBuilder, BATCH_BYTES_LIMIT, noop)
+    request = createHttpRequest([endpointBuilder], BATCH_BYTES_LIMIT, noop)
   })
 
   describe('send', () => {
@@ -90,6 +90,24 @@ describe('httpRequest', () => {
       await interceptor.waitForAllFetchCalls()
       await collectAsyncCalls(fetchSpy, 2)
     })
+
+    it('sends the payload to multiple endpoints', async () => {
+      const endpointBuilder2 = mockEndpointBuilder('http://my.website2')
+
+      request = createHttpRequest([endpointBuilder, endpointBuilder2], BATCH_BYTES_LIMIT, noop)
+
+      interceptor.withFetch(DEFAULT_FETCH_MOCK, DEFAULT_FETCH_MOCK)
+
+      const payloadData = '{"foo":"bar1"}\n{"foo":"bar2"}'
+      request.send({ data: payloadData, bytesCount: 10 })
+
+      await interceptor.waitForAllFetchCalls()
+      expect(requests.length).toEqual(2)
+      expect(requests[0].url).toContain('http://my.website')
+      expect(requests[0].body).toEqual(payloadData)
+      expect(requests[1].url).toContain('http://my.website2')
+      expect(requests[1].body).toEqual(payloadData)
+    })
   })
 
   describe('fetchKeepAliveStrategy onResponse', () => {
@@ -313,7 +331,7 @@ describe('httpRequest intake parameters', () => {
     interceptor = interceptRequests()
     requests = interceptor.requests
     endpointBuilder = createEndpointBuilder({ clientToken }, 'logs')
-    request = createHttpRequest(endpointBuilder, BATCH_BYTES_LIMIT, noop)
+    request = createHttpRequest([endpointBuilder], BATCH_BYTES_LIMIT, noop)
   })
 
   it('should have a unique request id', async () => {

packages/core/src/transport/httpRequest.ts

@@ -63,33 +63,35 @@ export interface RetryInfo {
 }
 
 export function createHttpRequest<Body extends Payload = Payload>(
-  endpointBuilder: EndpointBuilder,
+  endpointBuilders: EndpointBuilder[],
   bytesLimit: number,
   reportError: (error: RawError) => void
 ): HttpRequest<Body> {
   const observable = new Observable<HttpRequestEvent<Body>>()
   const retryState = newRetryState<Body>()
-  const sendStrategyForRetry = (payload: Body, onResponse: (r: HttpResponse) => void) =>
-    fetchKeepAliveStrategy(endpointBuilder, bytesLimit, payload, onResponse)
 
   return {
     observable,
     send: (payload: Body) => {
-      sendWithRetryStrategy(
-        payload,
-        retryState,
-        sendStrategyForRetry,
-        endpointBuilder.trackType,
-        reportError,
-        observable
-      )
+      for (const endpointBuilder of endpointBuilders) {
+        sendWithRetryStrategy(
+          payload,
+          retryState,
+          (payload, onResponse) => fetchKeepAliveStrategy(endpointBuilder, bytesLimit, payload, onResponse),
+          endpointBuilder.trackType,
+          reportError,
+          observable
+        )
+      }
     },
     /**
      * Since fetch keepalive behaves like regular fetch on Firefox,
      * keep using sendBeaconStrategy on exit
      */
     sendOnExit: (payload: Body) => {
-      sendBeaconStrategy(endpointBuilder, bytesLimit, payload)
+      for (const endpointBuilder of endpointBuilders) {
+        sendBeaconStrategy(endpointBuilder, bytesLimit, payload)
+      }
     },
   }
 }