Add latency bucket calculation

src/storages/__tests__/findLatencyIndex.spec.ts

@@ -0,0 +1,27 @@
+import { findLatencyIndex } from '../findLatencyIndex';
+
+const latenciesInMsAndBuckets = [
+  // First bucket is up to 0.5 ms
+  [0, 0],
+  [0.500, 0],
+  [1.400, 1],
+  [1.500, 1],
+  [8.000, 6],
+  [11.39, 6],
+  [11.392, 7],
+  [17.085, 7],
+  // Last bucket
+  [7481.827, 22],
+  [7481.828, 22],
+  [7999.999, 22],
+  // Invalid values
+  [NaN, 0]
+];
+
+test('findLatencyIndex', () => {
+
+  latenciesInMsAndBuckets.forEach(([latencyInMs, expectedBucket]) => { // @ts-ignore
+    expect(findLatencyIndex(latencyInMs)).toBe(expectedBucket);
+  });
+
+});

src/storages/findLatencyIndex.ts

@@ -1,7 +1,16 @@
 import { isNaNNumber } from '../utils/lang';
 
-// @TODO add unit tests
-export function findLatencyIndex(latency: number, min = 0, max = 23, base = 1.5): number {
-  const index = Math.min(max, Math.max(min, Math.floor(Math.log(latency) / Math.log(base))));
+const MIN = 0;
+const MAX = 22;
+const BASE = 1.5;
+
+/**
+ * Calculates buckets from latency in milliseconds
+ *
+ * @param latencyInMs
+ * @returns a bucket index from 0 to 22 inclusive
+ */
+export function findLatencyIndex(latencyInMs: number): number {
+  const index = Math.min(MAX, Math.max(MIN, Math.ceil(Math.log(latencyInMs) / Math.log(BASE))));
   return isNaNNumber(index) ? 0 : index; // index is NaN if latency is not a positive number
 }

src/storages/inMemory/TelemetryCacheInMemory.ts

@@ -1,10 +1,15 @@
 import { ImpressionDataType, EventDataType, LastSync, HttpErrors, HttpLatencies, StreamingEvent, Method, OperationType, MethodExceptions, MethodLatencies } from '../../sync/submitters/types';
+import { findLatencyIndex } from '../findLatencyIndex';
 import { TelemetryCacheSync } from '../types';
 
-const MAX_LATENCY_BUCKET_COUNT = 23;
 const MAX_STREAMING_EVENTS = 20;
 const MAX_TAGS = 10;
 
+function newBuckets() {
+  // MAX_LATENCY_BUCKET_COUNT (length) is 23;
+  return [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0];
+}
+
 export class TelemetryCacheInMemory implements TelemetryCacheSync {
 
   private timeUntilReady?: number;
@@ -77,7 +82,7 @@ export class TelemetryCacheInMemory implements TelemetryCacheSync {
     return result;
   }
 
-  recordSyncError(resource: OperationType, status: number) {
+  recordHttpError(resource: OperationType, status: number) {
     if (!this.httpErrors[resource]) this.httpErrors[resource] = {};
     if (!this.httpErrors[resource][status]) {
       this.httpErrors[resource][status] = 1;
@@ -95,11 +100,11 @@ export class TelemetryCacheInMemory implements TelemetryCacheSync {
     return result;
   }
 
-  recordSyncLatency(resource: OperationType, latencyMs: number) {
-    if (!this.httpLatencies[resource]) this.httpLatencies[resource] = [];
-    if (this.httpLatencies[resource].length < MAX_LATENCY_BUCKET_COUNT) {
-      this.httpLatencies[resource].push(latencyMs);
+  recordHttpLatency(resource: OperationType, latencyMs: number) {
+    if (!this.httpLatencies[resource]) {
+      this.httpLatencies[resource] = newBuckets();
     }
+    this.httpLatencies[resource][findLatencyIndex(latencyMs)]++;
   }
 
   private authRejections = 0;
@@ -187,10 +192,10 @@ export class TelemetryCacheInMemory implements TelemetryCacheSync {
   }
 
   recordLatency(method: Method, latencyMs: number) {
-    if (!this.latencies[method]) this.latencies[method] = [];
-    if (this.latencies[method].length < MAX_LATENCY_BUCKET_COUNT) {
-      this.latencies[method].push(latencyMs);
+    if (!this.latencies[method]) {
+      this.latencies[method] = newBuckets();
     }
+    this.latencies[method][findLatencyIndex(latencyMs)]++;
   }
 
 }

src/storages/inMemory/__tests__/TelemetryCacheInMemory.spec.ts

@@ -25,6 +25,9 @@ const methods: Method[] = [
   TRACK
 ];
 
+const latencies = [0, 0.500, 1.400, 17.085, 7999.999];
+const latencyBuckets = [2, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1];
+
 describe('TELEMETRY CACHE', () => {
   const cache = new TelemetryCacheInMemory();
 
@@ -92,9 +95,9 @@ describe('TELEMETRY CACHE', () => {
   test('http errors', () => {
     expect(cache.popHttpErrors()).toEqual({});
     operationTypes.forEach((operation) => {
-      cache.recordSyncError(operation, 400);
-      cache.recordSyncError(operation, 400);
-      cache.recordSyncError(operation, 500);
+      cache.recordHttpError(operation, 400);
+      cache.recordHttpError(operation, 400);
+      cache.recordHttpError(operation, 500);
     });
 
     const httpErrors = { '400': 2, '500': 1 };
@@ -103,28 +106,20 @@ describe('TELEMETRY CACHE', () => {
     expect(cache.popHttpErrors()).toEqual({});
 
     // Set a single http error
-    cache.recordSyncError(MY_SEGMENT, 400);
+    cache.recordHttpError(MY_SEGMENT, 400);
     expect(cache.popHttpErrors()).toEqual({ 'ms': { 400: 1 } });
   });
 
   test('http latencies', () => {
     expect(cache.popHttpLatencies()).toEqual({});
     operationTypes.forEach((operation) => {
-      cache.recordSyncLatency(operation, 300);
-      cache.recordSyncLatency(operation, 400);
-      cache.recordSyncLatency(operation, 500);
+      latencies.forEach((latency) => {
+        cache.recordHttpLatency(operation, latency);
+      });
     });
 
-    const latencies = [300, 400, 500];
-    const expectedLatencies = { 'sp': latencies, 'im': latencies, 'ic': latencies, 'ev': latencies, 'te': latencies, 'to': latencies, 'se': latencies, 'ms': latencies };
-    expect(cache.popHttpLatencies()).toEqual(expectedLatencies);
-    expect(cache.popHttpLatencies()).toEqual({});
-
-    // MAX_LATENCY_BUCKET_COUNT === 23
-    for (let i = 0; i < 100; i++) {
-      cache.recordSyncLatency(MY_SEGMENT, i);
-    }
-    expect(cache.popHttpLatencies()).toEqual({ 'ms': [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22] });
+    const expectedLatencyBuckets = { 'sp': latencyBuckets, 'im': latencyBuckets, 'ic': latencyBuckets, 'ev': latencyBuckets, 'te': latencyBuckets, 'to': latencyBuckets, 'se': latencyBuckets, 'ms': latencyBuckets };
+    expect(cache.popHttpLatencies()).toEqual(expectedLatencyBuckets);
     expect(cache.popHttpLatencies()).toEqual({});
   });
 
@@ -204,21 +199,13 @@ describe('TELEMETRY CACHE', () => {
   test('method latencies', () => {
     expect(cache.popLatencies()).toEqual({});
     methods.forEach((method) => {
-      cache.recordLatency(method, 300);
-      cache.recordLatency(method, 400);
-      cache.recordLatency(method, 500);
+      latencies.forEach((latency) => {
+        cache.recordLatency(method, latency);
+      });
     });
 
-    const latencies = [300, 400, 500];
-    const expectedLatencies = { 't': latencies, 'ts': latencies, 'tc': latencies, 'tcs': latencies, 'tr': latencies };
-    expect(cache.popLatencies()).toEqual(expectedLatencies);
-    expect(cache.popLatencies()).toEqual({});
-
-    // MAX_LATENCY_BUCKET_COUNT === 23
-    for (let i = 0; i < 100; i++) {
-      cache.recordLatency(TRACK, i);
-    }
-    expect(cache.popLatencies()).toEqual({ 'tr': [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22] });
+    const expectedLatencyBuckets = { 't': latencyBuckets, 'ts': latencyBuckets, 'tc': latencyBuckets, 'tcs': latencyBuckets, 'tr': latencyBuckets };
+    expect(cache.popLatencies()).toEqual(expectedLatencyBuckets);
     expect(cache.popLatencies()).toEqual({});
   });
 

src/storages/types.ts

@@ -398,8 +398,8 @@ export interface TelemetryRuntimeProducerSync {
   recordImpressionStats(type: ImpressionDataType, count: number): void;
   recordEventStats(type: EventDataType, count: number): void;
   recordSuccessfulSync(resource: OperationType, timeMs: number): void;
-  recordSyncError(resource: OperationType, status: number): void;
-  recordSyncLatency(resource: OperationType, latencyMs: number): void;
+  recordHttpError(resource: OperationType, status: number): void;
+  recordHttpLatency(resource: OperationType, latencyMs: number): void;
   recordAuthRejections(): void;
   recordTokenRefreshes(): void;
   recordStreamingEvents(streamingEvent: StreamingEvent): void;