Merge pull request #285 from splitio/fix_telemetry_in_consumer

Fix for telemetry in consumer mode

Author: EmilianoSanchez

CHANGES.txt

@@ -1,5 +1,6 @@
 1.12.1 (December 12, 2023)
  - Updated PluggableStorage for producer mode, to clear the storage if it was previously synchronized with a different SDK key (i.e., a different environment) or different Split Filter criteria.
+ - Bugfixing - Fixed an issue when tracking telemetry latencies for the new `getTreatmentsByFlagSet(s)` methods in Redis and Pluggable storages, which was causing the SDK to not track those stats.
 
 1.12.0 (December 4, 2023)
  - Added support for Flag Sets in "consumer" and "partial consumer" modes for Pluggable and Redis storages.

src/storages/inRedis/TelemetryCacheInRedis.ts

@@ -69,15 +69,10 @@ export class TelemetryCacheInRedis implements ITelemetryCacheAsync {
           return;
         }
 
-        if (!result.has(metadata)) result.set(metadata, {
-          t: newBuckets(),
-          ts: newBuckets(),
-          tc: newBuckets(),
-          tcs: newBuckets(),
-          tr: newBuckets(),
-        });
-
-        result.get(metadata)![method]![bucket] = count;
+        const methodLatencies = result.get(metadata) || {};
+        methodLatencies[method] = methodLatencies[method] || newBuckets();
+        methodLatencies[method]![bucket] = count;
+        result.set(metadata, methodLatencies);
       });
 
       return this.redis.del(this.keys.latencyPrefix).then(() => result);
@@ -109,14 +104,7 @@ export class TelemetryCacheInRedis implements ITelemetryCacheAsync {
 
         const [metadata, method] = parsedField;
 
-        if (!result.has(metadata)) result.set(metadata, {
-          t: 0,
-          ts: 0,
-          tc: 0,
-          tcs: 0,
-          tr: 0,
-        });
-
+        if (!result.has(metadata)) result.set(metadata, {});
         result.get(metadata)![method] = count;
       });
 

src/storages/inRedis/__tests__/TelemetryCacheInRedis.spec.ts

@@ -1,7 +1,6 @@
 import { loggerMock } from '../../../logger/__tests__/sdkLogger.mock';
 import { KeyBuilderSS } from '../../KeyBuilderSS';
 import { TelemetryCacheInRedis } from '../TelemetryCacheInRedis';
-import { newBuckets } from '../../inMemory/TelemetryCacheInMemory';
 import { metadata } from '../../__tests__/KeyBuilder.spec';
 import { RedisAdapter } from '../RedisAdapter';
 
@@ -20,16 +19,20 @@ test('TELEMETRY CACHE IN REDIS', async () => {
   // recordException
   expect(await cache.recordException('tr')).toBe(1);
   expect(await cache.recordException('tr')).toBe(2);
+  expect(await cache.recordException('tcfs')).toBe(1);
 
   expect(await connection.hget(exceptionKey, fieldVersionablePrefix + '/track')).toBe('2');
   expect(await connection.hget(exceptionKey, fieldVersionablePrefix + '/treatment')).toBe(null);
+  expect(await connection.hget(exceptionKey, fieldVersionablePrefix + '/treatmentsWithConfigByFlagSets')).toBe('1');
 
   // recordLatency
   expect(await cache.recordLatency('tr', 1.6)).toBe(1);
   expect(await cache.recordLatency('tr', 1.6)).toBe(2);
+  expect(await cache.recordLatency('tfs', 1.6)).toBe(1);
 
   expect(await connection.hget(latencyKey, fieldVersionablePrefix + '/track/2')).toBe('2');
   expect(await connection.hget(latencyKey, fieldVersionablePrefix + '/treatment/2')).toBe(null);
+  expect(await connection.hget(latencyKey, fieldVersionablePrefix + '/treatmentsByFlagSets/2')).toBe('1');
 
   // recordConfig
   expect(await cache.recordConfig()).toBe(1);
@@ -45,10 +48,7 @@ test('TELEMETRY CACHE IN REDIS', async () => {
   latencies.forEach((latency, m) => {
     expect(JSON.parse(m)).toEqual(metadata);
     expect(latency).toEqual({
-      t: newBuckets(),
-      ts: newBuckets(),
-      tc: newBuckets(),
-      tcs: newBuckets(),
+      tfs: [0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
       tr: [0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
     });
   });
@@ -59,10 +59,7 @@ test('TELEMETRY CACHE IN REDIS', async () => {
   exceptions.forEach((exception, m) => {
     expect(JSON.parse(m)).toEqual(metadata);
     expect(exception).toEqual({
-      t: 0,
-      ts: 0,
-      tc: 0,
-      tcs: 0,
+      tcfs: 1,
       tr: 2,
     });
   });

src/storages/pluggable/TelemetryCachePluggable.ts

@@ -68,15 +68,10 @@ export class TelemetryCachePluggable implements ITelemetryCacheAsync {
               continue;
             }
 
-            if (!result.has(metadata)) result.set(metadata, {
-              t: newBuckets(),
-              ts: newBuckets(),
-              tc: newBuckets(),
-              tcs: newBuckets(),
-              tr: newBuckets(),
-            });
-
-            result.get(metadata)![method]![bucket] = count;
+            const methodLatencies = result.get(metadata) || {};
+            methodLatencies[method] = methodLatencies[method] || newBuckets();
+            methodLatencies[method]![bucket] = count;
+            result.set(metadata, methodLatencies);
           }
 
           return Promise.all(latencyKeys.map((latencyKey) => this.wrapper.del(latencyKey))).then(() => result);
@@ -115,14 +110,7 @@ export class TelemetryCachePluggable implements ITelemetryCacheAsync {
 
             const [metadata, method] = parsedField;
 
-            if (!result.has(metadata)) result.set(metadata, {
-              t: 0,
-              ts: 0,
-              tc: 0,
-              tcs: 0,
-              tr: 0,
-            });
-
+            if (!result.has(metadata)) result.set(metadata, {});
             result.get(metadata)![method] = count;
           }
 

src/storages/pluggable/__tests__/TelemetryCachePluggable.spec.ts

@@ -19,16 +19,20 @@ test('TELEMETRY CACHE PLUGGABLE', async () => {
   // recordException
   expect(await cache.recordException('tr')).toBe(1);
   expect(await cache.recordException('tr')).toBe(2);
+  expect(await cache.recordException('tcfs')).toBe(1);
 
   expect(await wrapper.get(exceptionKey + '::' + fieldVersionablePrefix + '/track')).toBe('2');
   expect(await wrapper.get(exceptionKey + '::' + fieldVersionablePrefix + '/treatment')).toBe(null);
+  expect(await wrapper.get(exceptionKey + '::' + fieldVersionablePrefix + '/treatmentsWithConfigByFlagSets')).toBe('1');
 
   // recordLatency
   expect(await cache.recordLatency('tr', 1.6)).toBe(1);
   expect(await cache.recordLatency('tr', 1.6)).toBe(2);
+  expect(await cache.recordLatency('tfs', 1.6)).toBe(1);
 
   expect(await wrapper.get(latencyKey + '::' + fieldVersionablePrefix + '/track/2')).toBe('2');
   expect(await wrapper.get(latencyKey + '::' + fieldVersionablePrefix + '/treatment/2')).toBe(null);
+  expect(await wrapper.get(latencyKey + '::' + fieldVersionablePrefix + '/treatmentsByFlagSets/2')).toBe('1');
 
   // recordConfig
   await cache.recordConfig();
@@ -44,13 +48,15 @@ test('TELEMETRY CACHE PLUGGABLE', async () => {
   latencies.forEach((latency, m) => {
     expect(JSON.parse(m)).toEqual(metadata);
     expect(latency.tr![2]).toBe(2);
+    expect(latency.tfs![2]).toBe(1);
   });
 
   // popExceptions
   const exceptions = await cache.popExceptions();
   exceptions.forEach((exception, m) => {
     expect(JSON.parse(m)).toEqual(metadata);
     expect(exception.tr).toBe(2);
+    expect(exception.tcfs).toBe(1);
   });
 
   // popConfigs