fix: fix duplicative log payloads on unload (#1166)

Author: metal-messiah

src/common/constants/agent-constants.js

@@ -0,0 +1,2 @@
+export const IDEAL_PAYLOAD_SIZE = 64000
+export const MAX_PAYLOAD_SIZE = 1000000

src/features/logging/aggregate/index.js

@@ -8,10 +8,12 @@ import { warn } from '../../../common/util/console'
 import { stringify } from '../../../common/util/stringify'
 import { SUPPORTABILITY_METRIC_CHANNEL } from '../../metrics/constants'
 import { AggregateBase } from '../../utils/aggregate-base'
-import { FEATURE_NAME, LOGGING_EVENT_EMITTER_CHANNEL, LOG_LEVELS, MAX_PAYLOAD_SIZE } from '../constants'
+import { FEATURE_NAME, LOGGING_EVENT_EMITTER_CHANNEL, LOG_LEVELS } from '../constants'
 import { Log } from '../shared/log'
 import { isValidLogLevel } from '../shared/utils'
 import { applyFnToProps } from '../../../common/util/traverse'
+import { MAX_PAYLOAD_SIZE } from '../../../common/constants/agent-constants'
+import { EventBuffer } from '../../utils/event-buffer'
 
 export class Aggregate extends AggregateBase {
   static featureName = FEATURE_NAME
@@ -21,11 +23,7 @@ export class Aggregate extends AggregateBase {
     super(agentIdentifier, aggregator, FEATURE_NAME)
 
     /** held logs before sending */
-    this.bufferedLogs = []
-    /** held logs during sending, for retries */
-    this.outgoingLogs = []
-    /** the estimated bytes of log data waiting to be sent -- triggers a harvest if adding a new log will exceed limit  */
-    this.estimatedBytes = 0
+    this.bufferedLogs = new EventBuffer()
 
     this.#agentRuntime = getRuntime(this.agentIdentifier)
     this.#agentInfo = getInfo(this.agentIdentifier)
@@ -39,11 +37,11 @@ export class Aggregate extends AggregateBase {
         getPayload: this.prepareHarvest.bind(this),
         raw: true
       }, this)
-      /** harvest immediately once started to purge pre-load logs collected */
-      this.scheduler.startTimer(this.harvestTimeSeconds, 0)
       /** emitted by instrument class (wrapped loggers) or the api methods directly */
       registerHandler(LOGGING_EVENT_EMITTER_CHANNEL, this.handleLog.bind(this), this.featureName, this.ee)
       this.drain()
+      /** harvest immediately once started to purge pre-load logs collected */
+      this.scheduler.startTimer(this.harvestTimeSeconds, 0)
     })
   }
 
@@ -70,10 +68,6 @@ export class Aggregate extends AggregateBase {
       return
     }
     if (typeof message !== 'string' || !message) return warn(32)
-    if (message.length > MAX_PAYLOAD_SIZE) {
-      handle(SUPPORTABILITY_METRIC_CHANNEL, ['Logging/Harvest/Failed/Seen', message.length])
-      return warn(31, message.slice(0, 25) + '...')
-    }
 
     const log = new Log(
       Math.floor(this.#agentRuntime.timeKeeper.correctAbsoluteTimestamp(
@@ -84,26 +78,26 @@ export class Aggregate extends AggregateBase {
       level
     )
     const logBytes = log.message.length + stringify(log.attributes).length + log.level.length + 10 // timestamp == 10 chars
-    if (logBytes > MAX_PAYLOAD_SIZE) {
-      handle(SUPPORTABILITY_METRIC_CHANNEL, ['Logging/Harvest/Failed/Seen', logBytes])
-      return warn(31, log.message.slice(0, 25) + '...')
-    }
 
-    if (this.estimatedBytes + logBytes >= MAX_PAYLOAD_SIZE) {
-      handle(SUPPORTABILITY_METRIC_CHANNEL, ['Logging/Harvest/Early/Seen', this.estimatedBytes + logBytes])
-      this.scheduler.runHarvest({})
+    if (!this.bufferedLogs.canMerge(logBytes)) {
+      if (this.bufferedLogs.hasData) {
+        handle(SUPPORTABILITY_METRIC_CHANNEL, ['Logging/Harvest/Early/Seen', this.bufferedLogs.bytes + logBytes])
+        this.scheduler.runHarvest({})
+        if (logBytes < MAX_PAYLOAD_SIZE) this.bufferedLogs.add(log)
+      } else {
+        handle(SUPPORTABILITY_METRIC_CHANNEL, ['Logging/Harvest/Failed/Seen', logBytes])
+        warn(31, log.message.slice(0, 25) + '...')
+      }
+      return
     }
-    this.estimatedBytes += logBytes
-    this.bufferedLogs.push(log)
+
+    this.bufferedLogs.add(log)
   }
 
-  prepareHarvest () {
-    if (this.blocked || !(this.bufferedLogs.length || this.outgoingLogs.length)) return
-    /** populate outgoing array while also clearing main buffer */
-    this.outgoingLogs.push(...this.bufferedLogs.splice(0))
-    this.estimatedBytes = 0
+  prepareHarvest (options = {}) {
+    if (this.blocked || !this.bufferedLogs.hasData) return
     /** see https://source.datanerd.us/agents/rum-specs/blob/main/browser/Log for logging spec */
-    return {
+    const payload = {
       qs: {
         browser_monitoring_key: this.#agentInfo.licenseKey
       },
@@ -123,14 +117,20 @@ export class Aggregate extends AggregateBase {
         },
         /** logs section contains individual unique log entries */
         logs: applyFnToProps(
-          this.outgoingLogs,
+          this.bufferedLogs.buffer,
           this.obfuscator.obfuscateString.bind(this.obfuscator), 'string'
         )
       }]
     }
+
+    if (options.retry) this.bufferedLogs.hold()
+    else this.bufferedLogs.clear()
+
+    return payload
   }
 
   onHarvestFinished (result) {
-    if (!result.retry) this.outgoingLogs = []
+    if (result.retry) this.bufferedLogs.unhold()
+    else this.bufferedLogs.held.clear()
   }
 }

src/features/utils/event-buffer.js

@@ -0,0 +1,126 @@
+import { stringify } from '../../common/util/stringify'
+import { MAX_PAYLOAD_SIZE } from '../../common/constants/agent-constants'
+
+/**
+ * A container that keeps an event buffer and size with helper methods
+ * @typedef {Object} EventBuffer
+ * @property {number} size
+ * @property {*[]} buffer
+ */
+
+/**
+ * A container that holds, evaluates, and merges event objects for harvesting
+ */
+export class EventBuffer {
+  /** @type {Object[]} */
+  #buffer = []
+  /** @type {number} */
+  #bytes = 0
+  /** @type {EventBuffer} */
+  #held
+
+  /**
+   *
+   * @param {number=} maxPayloadSize
+   */
+  constructor (maxPayloadSize = MAX_PAYLOAD_SIZE) {
+    this.maxPayloadSize = maxPayloadSize
+  }
+
+  /**
+   * buffer is read only, use the helper methods to add or clear buffer data
+   */
+  get buffer () {
+    return this.#buffer
+  }
+
+  /**
+   * bytes is read only, use the helper methods to add or clear buffer data
+   */
+  get bytes () {
+    return this.#bytes
+  }
+
+  /**
+   * held is another event buffer
+   */
+  get held () {
+    this.#held ??= new EventBuffer(this.maxPayloadSize)
+    return this.#held
+  }
+
+  /**
+   * Returns a boolean indicating whether the current size and buffer contain valid data
+   * @returns {boolean}
+   */
+  get hasData () {
+    return this.buffer.length > 0 && this.bytes > 0
+  }
+
+  /**
+   * Adds an event object to the buffer while tallying size
+   * @param {Object} event the event object to add to the buffer
+   * @returns {EventBuffer} returns the event buffer for chaining
+   */
+  add (event) {
+    const size = stringify(event).length
+    if (!this.canMerge(size)) return this
+    this.#buffer.push(event)
+    this.#bytes += size
+    return this
+  }
+
+  /**
+   * clear the buffer data
+   * @returns {EventBuffer}
+   */
+  clear () {
+    this.#bytes = 0
+    this.#buffer = []
+    return this
+  }
+
+  /**
+   * Hold the buffer data in a new (child) EventBuffer (.held) to unblock the main buffer.
+   * This action clears the main buffer
+   * @returns {EventBuffer}
+   */
+  hold () {
+    this.held.merge(this)
+    this.clear()
+    return this
+  }
+
+  /**
+   * Prepend the held EventBuffer (.held) back into the main buffer
+   * This action clears the held buffer
+   * @returns {EventBuffer}
+   */
+  unhold () {
+    this.merge(this.held, true)
+    this.held.clear()
+    return this
+  }
+
+  /**
+   * Merges an EventBuffer into this EventBuffer
+   * @param {EventBuffer} events an EventBuffer intended to merge with this EventBuffer
+   * @param {boolean} prepend if true, the supplied events will be prepended before the events of this class
+   * @returns {EventBuffer} returns the event buffer for chaining
+   */
+  merge (eventBuffer, prepend = false) {
+    if (!this.canMerge(eventBuffer.bytes)) return this
+    this.#buffer = prepend ? [...eventBuffer.buffer, ...this.#buffer] : [...this.#buffer, ...eventBuffer.buffer]
+    this.#bytes += eventBuffer.#bytes
+    return this
+  }
+
+  /**
+   * Returns a boolean indicating the resulting size of a merge would be valid. Compares against the maxPayloadSize provided at initialization time.
+   * @param {number} size
+   * @returns {boolean}
+   */
+  canMerge (size) {
+    return this.bytes + (size || Infinity) < this.maxPayloadSize
+  }
+}

tests/assets/logs-redirect.html

@@ -0,0 +1,22 @@
+<!DOCTYPE html>
+<!--
+  Copyright 2020 New Relic Corporation.
+  PDX-License-Identifier: Apache-2.0
+-->
+<html>
+  <head>
+    <title>RUM Unit Test</title>
+    {init} {config} {loader}
+  </head>
+  <body>
+    Logs captured immediately followed by a redirect
+
+    <script>
+    window.addEventListener('load', () => setTimeout(() => redirectTo('https://gmail.com'), 500))
+      function redirectTo(url) {
+        window.newrelic.log(`redirect to ${url}`);
+        window.open(url, "_blank");
+      }
+    </script>
+  </body>
+</html>

tests/components/logging/aggregate.test.js

@@ -40,9 +40,7 @@ describe('class setup', () => {
       'featureName',
       'blocked',
       'bufferedLogs',
-      'outgoingLogs',
-      'harvestTimeSeconds',
-      'estimatedBytes'
+      'harvestTimeSeconds'
     ]))
   })
 
@@ -72,7 +70,7 @@ describe('payloads', () => {
       { myAttributes: 1 },
       'error'
     )
-    expect(loggingAggregate.bufferedLogs[0]).toEqual(expectedLog)
+    expect(loggingAggregate.bufferedLogs.buffer[0]).toEqual(expectedLog)
 
     expect(loggingAggregate.prepareHarvest()).toEqual({
       qs: { browser_monitoring_key: info.licenseKey },
@@ -97,7 +95,7 @@ describe('payloads', () => {
   test('prepares payload as expected', async () => {
     loggingAggregate.ee.emit(LOGGING_EVENT_EMITTER_CHANNEL, [1234, 'test message', { myAttributes: 1 }, 'error'])
 
-    expect(loggingAggregate.bufferedLogs[0]).toEqual(new Log(
+    expect(loggingAggregate.bufferedLogs.buffer[0]).toEqual(new Log(
       Math.floor(runtime.timeKeeper.correctAbsoluteTimestamp(
         runtime.timeKeeper.convertRelativeTimestamp(1234)
       )),
@@ -142,17 +140,19 @@ describe('payloads', () => {
       'error'
     )
 
+    const logs = loggingAggregate.bufferedLogs.buffer
+
     loggingAggregate.ee.emit(LOGGING_EVENT_EMITTER_CHANNEL, [1234, 'test message', [], 'ERROR'])
-    expect(loggingAggregate.bufferedLogs.pop()).toEqual(expected)
+    expect(logs.pop()).toEqual(expected)
 
     loggingAggregate.ee.emit(LOGGING_EVENT_EMITTER_CHANNEL, [1234, 'test message', true, 'ERROR'])
-    expect(loggingAggregate.bufferedLogs.pop()).toEqual(expected)
+    expect(logs.pop()).toEqual(expected)
 
     loggingAggregate.ee.emit(LOGGING_EVENT_EMITTER_CHANNEL, [1234, 'test message', 1, 'ERROR'])
-    expect(loggingAggregate.bufferedLogs.pop()).toEqual(expected)
+    expect(logs.pop()).toEqual(expected)
 
     loggingAggregate.ee.emit(LOGGING_EVENT_EMITTER_CHANNEL, [1234, 'test message', 'string', 'ERROR'])
-    expect(loggingAggregate.bufferedLogs.pop()).toEqual(expected)
+    expect(logs.pop()).toEqual(expected)
   })
 
   test('should work if log level is valid but wrong case', async () => {
@@ -166,18 +166,19 @@ describe('payloads', () => {
     )
 
     loggingAggregate.ee.emit(LOGGING_EVENT_EMITTER_CHANNEL, [1234, 'test message', {}, 'ErRoR'])
-    expect(loggingAggregate.bufferedLogs.pop()).toEqual(expected)
+    expect(loggingAggregate.bufferedLogs.buffer[0]).toEqual(expected)
   })
 
   test('should buffer logs with non-stringify-able message', async () => {
+    const logs = loggingAggregate.bufferedLogs.buffer
     loggingAggregate.ee.emit(LOGGING_EVENT_EMITTER_CHANNEL, [1234, new Error('test'), {}, 'error'])
-    expect(loggingAggregate.bufferedLogs.pop().message).toEqual('Error: test')
+    expect(logs.pop().message).toEqual('Error: test')
 
     loggingAggregate.ee.emit(LOGGING_EVENT_EMITTER_CHANNEL, [1234, new SyntaxError('test'), {}, 'error'])
-    expect(loggingAggregate.bufferedLogs.pop().message).toEqual('SyntaxError: test')
+    expect(logs.pop().message).toEqual('SyntaxError: test')
 
     loggingAggregate.ee.emit(LOGGING_EVENT_EMITTER_CHANNEL, [1234, Symbol('test'), {}, 'error'])
-    expect(loggingAggregate.bufferedLogs.pop().message).toEqual('Symbol(test)')
+    expect(logs.pop().message).toEqual('Symbol(test)')
   })
 })
 

tests/specs/logging/harvesting.e2e.js

@@ -77,6 +77,20 @@ describe('logging harvesting', () => {
       expect(JSON.parse(body)[0].logs[0].message).toEqual('Error: test')
     })
 
+    it('should not double harvest on navigation logs', async () => {
+      const [logsRequests] = await Promise.all([
+        logsCapture.waitForResult({ timeout: 15000 }),
+        browser.url(await browser.testHandle.assetURL('logs-redirect.html'))
+      ])
+
+      // 1 harvest
+      expect(logsRequests.length).toEqual(1)
+      const parsedBody = JSON.parse(logsRequests[0].request.body)
+      // 1 log in the 1 harvest
+      expect(parsedBody[0].logs.length).toEqual(1)
+      expect(parsedBody[0].logs[0].message).toEqual('redirect to https://gmail.com')
+    })
+
     it('should allow for re-wrapping and 3rd party wrapping', async () => {
       const [[{ request: { body } }]] = await Promise.all([
         logsCapture.waitForResult({ totalCount: 1 }),