diff --git a/.eslintrc b/.eslintrc index 6457f1b6..dd64c8f6 100644 --- a/.eslintrc +++ b/.eslintrc @@ -37,7 +37,8 @@ "no-unused-vars": "off", "@typescript-eslint/no-unused-vars": "error", "keyword-spacing": "error", - "comma-style": "error" + "comma-style": "error", + "no-trailing-spaces": "error" }, "overrides": [ diff --git a/package-lock.json b/package-lock.json index 9565303c..970149ae 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,6 +1,6 @@ { "name": "@splitsoftware/splitio-commons", - "version": "1.6.2-rc.10", + "version": "1.6.2-rc.13", "lockfileVersion": 1, "requires": true, "dependencies": { diff --git a/package.json b/package.json index c90c8445..629f309f 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@splitsoftware/splitio-commons", - "version": "1.6.2-rc.10", + "version": "1.6.2-rc.13", "description": "Split Javascript SDK common components", "main": "cjs/index.js", "module": "esm/index.js", @@ -24,6 +24,7 @@ "build:cjs": "rimraf cjs && tsc -m CommonJS --outDir cjs", "test": "jest", "test:coverage": "jest --coverage", + "all": "npm run check && npm run build && npm run test", "publish:rc": "npm run check && npm run test && npm run build && npm publish --tag rc", "publish:stable": "npm run check && npm run test && npm run build && npm publish" }, diff --git a/src/listeners/__tests__/browser.spec.ts b/src/listeners/__tests__/browser.spec.ts index 304ba8a5..7b7002fc 100644 --- a/src/listeners/__tests__/browser.spec.ts +++ b/src/listeners/__tests__/browser.spec.ts @@ -1,20 +1,8 @@ import { BrowserSignalListener } from '../browser'; -import { IEventsCacheSync, IImpressionCountsCacheSync, IImpressionsCacheSync, IStorageSync } from '../../storages/types'; +import { IEventsCacheSync, IImpressionCountsCacheSync, IImpressionsCacheSync, IStorageSync, ITelemetryCacheSync, IUniqueKeysCacheBase } from '../../storages/types'; import { ISplitApi } from '../../services/types'; import { fullSettings } from '../../utils/settingsValidation/__tests__/settings.mocks'; -jest.mock('../../sync/submitters/telemetrySubmitter', () => { - return { - telemetryCacheStatsAdapter: () => { - return { - isEmpty: () => false, - clear: () => { }, - pop: () => ({}), - }; - } - }; -}); - /* Mocks start */ const fakeImpression = { @@ -37,31 +25,42 @@ const fakeEvent = { const fakeImpressionCounts = { 'someFeature::0': 1 }; +const fakeUniqueKeys = { + keys: [{ k: 'emi', fs: ['split1'] }], +}; // Storage with impressionsCount and telemetry cache const fakeStorageOptimized = { // @ts-expect-error impressions: { isEmpty: jest.fn(), - clear: jest.fn(), pop() { return [fakeImpression]; } } as IImpressionsCacheSync, // @ts-expect-error events: { isEmpty: jest.fn(), - clear: jest.fn(), pop() { return [fakeEvent]; } } as IEventsCacheSync, // @ts-expect-error impressionCounts: { isEmpty: jest.fn(), - clear: jest.fn(), pop() { return fakeImpressionCounts; } - } as IImpressionCountsCacheSync, - telemetry: {} + } as IImpressionCountsCacheSync, // @ts-expect-error + uniqueKeys: { + isEmpty: jest.fn(), + pop() { + return fakeUniqueKeys; + } + } as IUniqueKeysCacheBase, // @ts-expect-error + telemetry: { + isEmpty: jest.fn(), + pop() { + return 'fake telemetry'; + } + } as ITelemetryCacheSync }; const fakeStorageDebug = { @@ -75,6 +74,7 @@ const fakeSplitApi = { postEventsBulk: jest.fn(() => Promise.resolve()), postTestImpressionsCount: jest.fn(() => Promise.resolve()), postMetricsUsage: jest.fn(() => Promise.resolve()), + postUniqueKeysBulkCs: jest.fn(() => Promise.resolve()), } as ISplitApi; const VISIBILITYCHANGE_EVENT = 'visibilitychange'; @@ -190,8 +190,8 @@ test('Browser JS listener / standalone mode / Impressions optimized mode with te triggerEvent(VISIBILITYCHANGE_EVENT, 'hidden'); - // Visibility change event was triggered. Thus sendBeacon method should be called four times. - expect(global.window.navigator.sendBeacon).toBeCalledTimes(4); + // Visibility change event was triggered. Thus sendBeacon method should be called five times (events, impressions, impression count, unique keys and telemetry). + expect(global.window.navigator.sendBeacon).toBeCalledTimes(5); // Http post services should have not been called expect(fakeSplitApi.postTestImpressionsBulk).not.toBeCalled(); @@ -297,8 +297,8 @@ test('Browser JS listener / standalone mode / user consent status', () => { settings.userConsent = undefined; triggerEvent(VISIBILITYCHANGE_EVENT, 'hidden'); - // Unload event was triggered when user consent was granted and undefined. Thus sendBeacon should be called 8 times (4 times per event in optimized mode with telemetry). - expect(global.window.navigator.sendBeacon).toBeCalledTimes(8); + // Unload event was triggered when user consent was granted and undefined. Thus sendBeacon should be called 10 times (5 times per event). + expect(global.window.navigator.sendBeacon).toBeCalledTimes(10); listener.stop(); }); diff --git a/src/listeners/browser.ts b/src/listeners/browser.ts index 11085f79..e30e0b26 100644 --- a/src/listeners/browser.ts +++ b/src/listeners/browser.ts @@ -1,7 +1,7 @@ /* eslint-disable no-undef */ // @TODO eventually migrate to JS-Browser-SDK package. import { ISignalListener } from './types'; -import { IRecorderCacheProducerSync, IStorageSync } from '../storages/types'; +import { IRecorderCacheSync, IStorageSync } from '../storages/types'; import { fromImpressionsCollector } from '../sync/submitters/impressionsSubmitter'; import { fromImpressionCountsCollector } from '../sync/submitters/impressionCountsSubmitter'; import { IResponse, ISplitApi } from '../services/types'; @@ -12,7 +12,6 @@ import { objectAssign } from '../utils/lang/objectAssign'; import { CLEANUP_REGISTERING, CLEANUP_DEREGISTERING } from '../logger/constants'; import { ISyncManager } from '../sync/types'; import { isConsentGranted } from '../consent'; -import { telemetryCacheStatsAdapter } from '../sync/submitters/telemetrySubmitter'; const VISIBILITYCHANGE_EVENT = 'visibilitychange'; const PAGEHIDE_EVENT = 'pagehide'; @@ -84,27 +83,25 @@ export class BrowserSignalListener implements ISignalListener { */ flushData() { if (!this.syncManager) return; // In consumer mode there is not sync manager and data to flush + const { events, telemetry } = this.settings.urls; // Flush impressions & events data if there is user consent if (isConsentGranted(this.settings)) { - const eventsUrl = this.settings.urls.events; const sim = this.settings.sync.impressionsMode; const extraMetadata = { // sim stands for Sync/Split Impressions Mode sim: sim === OPTIMIZED ? OPTIMIZED : sim === DEBUG ? DEBUG : NONE }; - this._flushData(eventsUrl + '/testImpressions/beacon', this.storage.impressions, this.serviceApi.postTestImpressionsBulk, this.fromImpressionsCollector, extraMetadata); - this._flushData(eventsUrl + '/events/beacon', this.storage.events, this.serviceApi.postEventsBulk); - if (this.storage.impressionCounts) this._flushData(eventsUrl + '/testImpressions/count/beacon', this.storage.impressionCounts, this.serviceApi.postTestImpressionsCount, fromImpressionCountsCollector); + this._flushData(events + '/testImpressions/beacon', this.storage.impressions, this.serviceApi.postTestImpressionsBulk, this.fromImpressionsCollector, extraMetadata); + this._flushData(events + '/events/beacon', this.storage.events, this.serviceApi.postEventsBulk); + if (this.storage.impressionCounts) this._flushData(events + '/testImpressions/count/beacon', this.storage.impressionCounts, this.serviceApi.postTestImpressionsCount, fromImpressionCountsCollector); + // @ts-ignore + if (this.storage.uniqueKeys) this._flushData(telemetry + '/v1/keys/cs/beacon', this.storage.uniqueKeys, this.serviceApi.postUniqueKeysBulkCs); } // Flush telemetry data - if (this.storage.telemetry) { - const telemetryUrl = this.settings.urls.telemetry; - const telemetryCacheAdapter = telemetryCacheStatsAdapter(this.storage.telemetry, this.storage.splits, this.storage.segments); - this._flushData(telemetryUrl + '/v1/metrics/usage/beacon', telemetryCacheAdapter, this.serviceApi.postMetricsUsage); - } + if (this.storage.telemetry) this._flushData(telemetry + '/v1/metrics/usage/beacon', this.storage.telemetry, this.serviceApi.postMetricsUsage); } flushDataIfHidden() { @@ -112,14 +109,13 @@ export class BrowserSignalListener implements ISignalListener { if (document.visibilityState === 'hidden') this.flushData(); // On a 'visibilitychange' event, flush data if state is hidden } - private _flushData(url: string, cache: IRecorderCacheProducerSync, postService: (body: string) => Promise, fromCacheToPayload?: (cacheData: T) => any, extraMetadata?: {}) { + private _flushData(url: string, cache: IRecorderCacheSync, postService: (body: string) => Promise, fromCacheToPayload?: (cacheData: T) => any, extraMetadata?: {}) { // if there is data in cache, send it to backend if (!cache.isEmpty()) { const dataPayload = fromCacheToPayload ? fromCacheToPayload(cache.pop()) : cache.pop(); if (!this._sendBeacon(url, dataPayload, extraMetadata)) { postService(JSON.stringify(dataPayload)).catch(() => { }); // no-op just to catch a possible exception } - cache.clear(); } } diff --git a/src/sdkClient/sdkClient.ts b/src/sdkClient/sdkClient.ts index 608d7bc3..36377b41 100644 --- a/src/sdkClient/sdkClient.ts +++ b/src/sdkClient/sdkClient.ts @@ -39,7 +39,7 @@ export function sdkClientFactory(params: ISdkFactoryContext, isSharedClient?: bo // Release the API Key if it is the main client if (!isSharedClient) releaseApiKey(settings.core.authorizationKey); - + if (uniqueKeysTracker) uniqueKeysTracker.stop(); // Cleanup storage diff --git a/src/sdkFactory/index.ts b/src/sdkFactory/index.ts index 3080fed6..35d53fca 100644 --- a/src/sdkFactory/index.ts +++ b/src/sdkFactory/index.ts @@ -3,14 +3,10 @@ import { sdkReadinessManagerFactory } from '../readiness/sdkReadinessManager'; import { impressionsTrackerFactory } from '../trackers/impressionsTracker'; import { eventTrackerFactory } from '../trackers/eventTracker'; import { telemetryTrackerFactory } from '../trackers/telemetryTracker'; -import { IStorageFactoryParams } from '../storages/types'; import { SplitIO } from '../types'; -import { getMatching } from '../utils/key'; -import { shouldBeOptimized } from '../trackers/impressionObserver/utils'; import { validateAndTrackApiKey } from '../utils/inputValidation/apiKey'; import { createLoggerAPI } from '../logger/sdkLogger'; import { NEW_FACTORY, RETRIEVE_MANAGER } from '../logger/constants'; -import { metadataBuilder } from '../storages/metadataBuilder'; import { SDK_SPLITS_ARRIVED, SDK_SEGMENTS_ARRIVED } from '../readiness/constants'; import { objectAssign } from '../utils/lang/objectAssign'; import { strategyDebugFactory } from '../trackers/strategy/strategyDebug'; @@ -28,7 +24,7 @@ export function sdkFactory(params: ISdkFactoryParams): SplitIO.ICsSDK | SplitIO. syncManagerFactory, SignalListener, impressionsObserverFactory, integrationsManagerFactory, sdkManagerFactory, sdkClientMethodFactory, filterAdapterFactory } = params; - const log = settings.log; + const { log, sync: { impressionsMode } } = settings; // @TODO handle non-recoverable errors, such as, global `fetch` not available, invalid API Key, etc. // On non-recoverable errors, we should mark the SDK as destroyed and not start synchronization. @@ -39,42 +35,24 @@ export function sdkFactory(params: ISdkFactoryParams): SplitIO.ICsSDK | SplitIO. const sdkReadinessManager = sdkReadinessManagerFactory(log, platform.EventEmitter, settings.startup.readyTimeout); const readiness = sdkReadinessManager.readinessManager; - // @TODO consider passing the settings object, so that each storage access only what it needs - const storageFactoryParams: IStorageFactoryParams = { - impressionsQueueSize: settings.scheduler.impressionsQueueSize, - eventsQueueSize: settings.scheduler.eventsQueueSize, - optimize: shouldBeOptimized(settings), - - // ATM, only used by InLocalStorage - matchingKey: getMatching(settings.core.key), - splitFiltersValidation: settings.sync.__splitFiltersValidation, - - // ATM, only used by PluggableStorage - mode: settings.mode, - impressionsMode: settings.sync.impressionsMode, - - // Callback used to emit SDK_READY in consumer mode, where `syncManagerFactory` is undefined, - // or partial consumer mode, where it only has submitters, and therefore it doesn't emit readiness events. + const storage = storageFactory({ + settings, onReadyCb: (error) => { if (error) return; // Don't emit SDK_READY if storage failed to connect. Error message is logged by wrapperAdapter readiness.splits.emit(SDK_SPLITS_ARRIVED); readiness.segments.emit(SDK_SEGMENTS_ARRIVED); }, - metadata: metadataBuilder(settings), - log - }; - - const storage = storageFactory(storageFactoryParams); + }); // @TODO add support for dataloader: `if (params.dataLoader) params.dataLoader(storage);` const telemetryTracker = telemetryTrackerFactory(storage.telemetry, platform.now); const integrationsManager = integrationsManagerFactory && integrationsManagerFactory({ settings, storage, telemetryTracker }); const observer = impressionsObserverFactory(); - const uniqueKeysTracker = storageFactoryParams.impressionsMode === NONE ? uniqueKeysTrackerFactory(log, storage.uniqueKeys!, filterAdapterFactory && filterAdapterFactory()) : undefined; + const uniqueKeysTracker = impressionsMode === NONE ? uniqueKeysTrackerFactory(log, storage.uniqueKeys!, filterAdapterFactory && filterAdapterFactory()) : undefined; let strategy; - switch (storageFactoryParams.impressionsMode) { + switch (impressionsMode) { case OPTIMIZED: strategy = strategyOptimizedFactory(observer, storage.impressionCounts!); break; @@ -120,7 +98,7 @@ export function sdkFactory(params: ISdkFactoryParams): SplitIO.ICsSDK | SplitIO. }, // Logger wrapper API - Logger: createLoggerAPI(settings.log), + Logger: createLoggerAPI(log), settings, }, extraProps && extraProps(ctx)); diff --git a/src/sdkFactory/types.ts b/src/sdkFactory/types.ts index c6937bc3..3cce9c9f 100644 --- a/src/sdkFactory/types.ts +++ b/src/sdkFactory/types.ts @@ -97,9 +97,9 @@ export interface ISdkFactoryParams { // It Allows to distinguish SDK clients with the client-side API (`ICsSDK`) or server-side API (`ISDK` or `IAsyncSDK`). sdkClientMethodFactory: (params: ISdkFactoryContext) => ({ (): SplitIO.ICsClient; (key: SplitIO.SplitKey, trafficType?: string | undefined): SplitIO.ICsClient; } | (() => SplitIO.IClient) | (() => SplitIO.IAsyncClient)) - // Impression observer factory. If provided, will be used for impressions dedupe + // Impression observer factory. impressionsObserverFactory: () => IImpressionObserver - + filterAdapterFactory?: () => IFilterAdapter // Optional signal listener constructor. Used to handle special app states, like shutdown, app paused or resumed. diff --git a/src/services/splitApi.ts b/src/services/splitApi.ts index 200b6bf4..9318992c 100644 --- a/src/services/splitApi.ts +++ b/src/services/splitApi.ts @@ -106,7 +106,7 @@ export function splitApiFactory( const url = `${urls.events}/testImpressions/count`; return splitHttpClient(url, { method: 'POST', body, headers }, telemetryTracker.trackHttp(IMPRESSIONS_COUNT)); }, - + /** * Post unique keys for client side. * @@ -117,7 +117,7 @@ export function splitApiFactory( const url = `${urls.telemetry}/v1/keys/cs`; return splitHttpClient(url, { method: 'POST', body, headers }, telemetryTracker.trackHttp(TELEMETRY)); }, - + /** * Post unique keys for server side. * diff --git a/src/storages/KeyBuilderSS.ts b/src/storages/KeyBuilderSS.ts index 84b8a62b..526b2628 100644 --- a/src/storages/KeyBuilderSS.ts +++ b/src/storages/KeyBuilderSS.ts @@ -1,9 +1,8 @@ import { KeyBuilder } from './KeyBuilder'; import { IMetadata } from '../dtos/types'; import { Method } from '../sync/submitters/types'; -import { MAX_LATENCY_BUCKET_COUNT } from './inMemory/TelemetryCacheInMemory'; -const METHOD_NAMES: Record = { +export const METHOD_NAMES: Record = { t: 'treatment', ts: 'treatments', tc: 'treatmentWithConfig', @@ -37,7 +36,7 @@ export class KeyBuilderSS extends KeyBuilder { buildImpressionsCountKey() { return `${this.prefix}.impressions.count`; } - + buildUniqueKeysKey() { return `${this.prefix}.uniquekeys`; } @@ -65,44 +64,3 @@ export class KeyBuilderSS extends KeyBuilder { } } - -// Used by consumer methods of TelemetryCacheInRedis and TelemetryCachePluggable - -const REVERSE_METHOD_NAMES = Object.keys(METHOD_NAMES).reduce((acc, key) => { - acc[METHOD_NAMES[key as Method]] = key as Method; - return acc; -}, {} as Record); - - -export function parseMetadata(field: string): [metadata: string] | string { - const parts = field.split('/'); - if (parts.length !== 3) return `invalid subsection count. Expected 3, got: ${parts.length}`; - - const [s /* metadata.s */, n /* metadata.n */, i /* metadata.i */] = parts; - return [JSON.stringify({ s, n, i })]; -} - -export function parseExceptionField(field: string): [metadata: string, method: Method] | string { - const parts = field.split('/'); - if (parts.length !== 4) return `invalid subsection count. Expected 4, got: ${parts.length}`; - - const [s /* metadata.s */, n /* metadata.n */, i /* metadata.i */, m] = parts; - const method = REVERSE_METHOD_NAMES[m]; - if (!method) return `unknown method '${m}'`; - - return [JSON.stringify({ s, n, i }), method]; -} - -export function parseLatencyField(field: string): [metadata: string, method: Method, bucket: number] | string { - const parts = field.split('/'); - if (parts.length !== 5) return `invalid subsection count. Expected 5, got: ${parts.length}`; - - const [s /* metadata.s */, n /* metadata.n */, i /* metadata.i */, m, b] = parts; - const method = REVERSE_METHOD_NAMES[m]; - if (!method) return `unknown method '${m}'`; - - const bucket = parseInt(b); - if (isNaN(bucket) || bucket >= MAX_LATENCY_BUCKET_COUNT) return `invalid bucket. Expected a number between 0 and ${MAX_LATENCY_BUCKET_COUNT - 1}, got: ${b}`; - - return [JSON.stringify({ s, n, i }), method, bucket]; -} diff --git a/src/storages/__tests__/findLatencyIndex.spec.ts b/src/storages/__tests__/findLatencyIndex.spec.ts index b2b6f04c..f99890d2 100644 --- a/src/storages/__tests__/findLatencyIndex.spec.ts +++ b/src/storages/__tests__/findLatencyIndex.spec.ts @@ -4,8 +4,11 @@ const latenciesInMsAndBuckets = [ // First bucket is up to 0.5 ms [0, 0], [0.500, 0], + [1.000, 0], + [1.001, 1], [1.400, 1], [1.500, 1], + [1.501, 2], [8.000, 6], [11.39, 6], [11.392, 7], diff --git a/src/storages/__tests__/testUtils.ts b/src/storages/__tests__/testUtils.ts index 203e2654..2f8cb245 100644 --- a/src/storages/__tests__/testUtils.ts +++ b/src/storages/__tests__/testUtils.ts @@ -10,6 +10,7 @@ export function assertStorageInterface(storage: IStorageSync | IStorageAsync) { expect(typeof storage.events).toBe('object'); expect(!storage.telemetry || typeof storage.telemetry === 'object').toBeTruthy; expect(!storage.impressionCounts || typeof storage.impressionCounts === 'object').toBeTruthy; + expect(!storage.uniqueKeys || typeof storage.uniqueKeys === 'object').toBeTruthy; } export function assertSyncRecorderCacheInterface(cache: IEventsCacheSync | IImpressionsCacheSync) { diff --git a/src/storages/__tests__/metadataBuilder.spec.ts b/src/storages/__tests__/utils.spec.ts similarity index 95% rename from src/storages/__tests__/metadataBuilder.spec.ts rename to src/storages/__tests__/utils.spec.ts index c3365308..c212a695 100644 --- a/src/storages/__tests__/metadataBuilder.spec.ts +++ b/src/storages/__tests__/utils.spec.ts @@ -1,4 +1,4 @@ -import { metadataBuilder } from '../metadataBuilder'; +import { metadataBuilder } from '../utils'; import { UNKNOWN } from '../../utils/constants'; describe('metadataBuilder', () => { diff --git a/src/storages/inLocalStorage/__tests__/index.spec.ts b/src/storages/inLocalStorage/__tests__/index.spec.ts index 417c8727..b9d0fc1d 100644 --- a/src/storages/inLocalStorage/__tests__/index.spec.ts +++ b/src/storages/inLocalStorage/__tests__/index.spec.ts @@ -1,7 +1,4 @@ // Mocks -import { loggerMock } from '../../../logger/__tests__/sdkLogger.mock'; -import { IStorageFactoryParams } from '../../types'; -import { assertStorageInterface } from '../../__tests__/testUtils'; const fakeInMemoryStorage = 'fakeStorage'; const fakeInMemoryStorageFactory = jest.fn(() => fakeInMemoryStorage); jest.mock('../../inMemory/InMemoryStorageCS', () => { @@ -10,13 +7,17 @@ jest.mock('../../inMemory/InMemoryStorageCS', () => { }; }); +import { IStorageFactoryParams } from '../../types'; +import { assertStorageInterface } from '../../__tests__/testUtils'; +import { fullSettings } from '../../../utils/settingsValidation/__tests__/settings.mocks'; + // Test target import { InLocalStorage } from '../index'; describe('IN LOCAL STORAGE', () => { // @ts-ignore - const internalSdkParams: IStorageFactoryParams = { log: loggerMock }; + const internalSdkParams: IStorageFactoryParams = { settings: fullSettings }; afterEach(() => { fakeInMemoryStorageFactory.mockClear(); diff --git a/src/storages/inLocalStorage/index.ts b/src/storages/inLocalStorage/index.ts index b7c721a9..59eefae5 100644 --- a/src/storages/inLocalStorage/index.ts +++ b/src/storages/inLocalStorage/index.ts @@ -12,9 +12,10 @@ import { SplitsCacheInMemory } from '../inMemory/SplitsCacheInMemory'; import { DEFAULT_CACHE_EXPIRATION_IN_MILLIS } from '../../utils/constants/browser'; import { InMemoryStorageCSFactory } from '../inMemory/InMemoryStorageCS'; import { LOG_PREFIX } from './constants'; -import { NONE, STORAGE_LOCALSTORAGE } from '../../utils/constants'; +import { DEBUG, NONE, STORAGE_LOCALSTORAGE } from '../../utils/constants'; import { shouldRecordTelemetry, TelemetryCacheInMemory } from '../inMemory/TelemetryCacheInMemory'; import { UniqueKeysCacheInMemoryCS } from '../inMemory/UniqueKeysCacheInMemoryCS'; +import { getMatching } from '../../utils/key'; export interface InLocalStorageOptions { prefix?: string @@ -31,22 +32,26 @@ export function InLocalStorage(options: InLocalStorageOptions = {}): IStorageSyn // Fallback to InMemoryStorage if LocalStorage API is not available if (!isLocalStorageAvailable()) { - params.log.warn(LOG_PREFIX + 'LocalStorage API is unavailable. Falling back to default MEMORY storage'); + params.settings.log.warn(LOG_PREFIX + 'LocalStorage API is unavailable. Falling back to default MEMORY storage'); return InMemoryStorageCSFactory(params); } - const log = params.log; - const keys = new KeyBuilderCS(prefix, params.matchingKey as string); + const { settings, settings: { log, scheduler: { impressionsQueueSize, eventsQueueSize, }, sync: { impressionsMode, __splitFiltersValidation } } } = params; + const matchingKey = getMatching(settings.core.key); + const keys = new KeyBuilderCS(prefix, matchingKey as string); const expirationTimestamp = Date.now() - DEFAULT_CACHE_EXPIRATION_IN_MILLIS; + const splits = new SplitsCacheInLocal(log, keys, expirationTimestamp, __splitFiltersValidation); + const segments = new MySegmentsCacheInLocal(log, keys); + return { - splits: new SplitsCacheInLocal(log, keys, expirationTimestamp, params.splitFiltersValidation), - segments: new MySegmentsCacheInLocal(log, keys), - impressions: new ImpressionsCacheInMemory(params.impressionsQueueSize), - impressionCounts: params.optimize ? new ImpressionCountsCacheInMemory() : undefined, - events: new EventsCacheInMemory(params.eventsQueueSize), - telemetry: shouldRecordTelemetry(params) ? new TelemetryCacheInMemory() : undefined, - uniqueKeys: params.impressionsMode === NONE ? new UniqueKeysCacheInMemoryCS() : undefined, + splits, + segments, + impressions: new ImpressionsCacheInMemory(impressionsQueueSize), + impressionCounts: impressionsMode !== DEBUG ? new ImpressionCountsCacheInMemory() : undefined, + events: new EventsCacheInMemory(eventsQueueSize), + telemetry: shouldRecordTelemetry(params) ? new TelemetryCacheInMemory(splits, segments) : undefined, + uniqueKeys: impressionsMode === NONE ? new UniqueKeysCacheInMemoryCS() : undefined, destroy() { this.splits = new SplitsCacheInMemory(); diff --git a/src/storages/inMemory/AttributesCacheInMemory.ts b/src/storages/inMemory/AttributesCacheInMemory.ts index 455d35e1..0718294a 100644 --- a/src/storages/inMemory/AttributesCacheInMemory.ts +++ b/src/storages/inMemory/AttributesCacheInMemory.ts @@ -7,10 +7,10 @@ export class AttributesCacheInMemory { /** * Create or update the value for the given attribute - * + * * @param {string} attributeName attribute name * @param {Object} attributeValue attribute value - * @returns {boolean} the attribute was stored + * @returns {boolean} the attribute was stored */ setAttribute(attributeName: string, attributeValue: Object): boolean { this.attributesCache[attributeName] = attributeValue; @@ -19,7 +19,7 @@ export class AttributesCacheInMemory { /** * Retrieves the value of a given attribute - * + * * @param {string} attributeName attribute name * @returns {Object?} stored attribute value */ @@ -29,7 +29,7 @@ export class AttributesCacheInMemory { /** * Create or update all the given attributes - * + * * @param {[string, Object]} attributes attributes to create or update * @returns {boolean} attributes were stored */ @@ -40,7 +40,7 @@ export class AttributesCacheInMemory { /** * Retrieve the full attributes map - * + * * @returns {Map} stored attributes */ getAll(): Record { @@ -49,7 +49,7 @@ export class AttributesCacheInMemory { /** * Removes a given attribute from the map - * + * * @param {string} attributeName attribute to remove * @returns {boolean} attribute removed */ @@ -63,7 +63,7 @@ export class AttributesCacheInMemory { /** * Clears all attributes stored in the SDK - * + * */ clear() { this.attributesCache = {}; diff --git a/src/storages/inMemory/InMemoryStorage.ts b/src/storages/inMemory/InMemoryStorage.ts index 678f903e..e576601c 100644 --- a/src/storages/inMemory/InMemoryStorage.ts +++ b/src/storages/inMemory/InMemoryStorage.ts @@ -14,15 +14,19 @@ import { UniqueKeysCacheInMemory } from './UniqueKeysCacheInMemory'; * @param params parameters required by EventsCacheSync */ export function InMemoryStorageFactory(params: IStorageFactoryParams): IStorageSync { + const { settings: { scheduler: { impressionsQueueSize, eventsQueueSize, }, sync: { impressionsMode } } } = params; + + const splits = new SplitsCacheInMemory(); + const segments = new SegmentsCacheInMemory(); return { - splits: new SplitsCacheInMemory(), - segments: new SegmentsCacheInMemory(), - impressions: new ImpressionsCacheInMemory(params.impressionsQueueSize), - impressionCounts: params.impressionsMode !== DEBUG ? new ImpressionCountsCacheInMemory() : undefined, - events: new EventsCacheInMemory(params.eventsQueueSize), - telemetry: shouldRecordTelemetry(params) ? new TelemetryCacheInMemory() : undefined, - uniqueKeys: params.impressionsMode === NONE ? new UniqueKeysCacheInMemory() : undefined, + splits, + segments, + impressions: new ImpressionsCacheInMemory(impressionsQueueSize), + impressionCounts: impressionsMode !== DEBUG ? new ImpressionCountsCacheInMemory() : undefined, + events: new EventsCacheInMemory(eventsQueueSize), + telemetry: shouldRecordTelemetry(params) ? new TelemetryCacheInMemory(splits, segments) : undefined, + uniqueKeys: impressionsMode === NONE ? new UniqueKeysCacheInMemory() : undefined, // When using MEMORY we should clean all the caches to leave them empty destroy() { diff --git a/src/storages/inMemory/InMemoryStorageCS.ts b/src/storages/inMemory/InMemoryStorageCS.ts index c20b6a93..6be93b7f 100644 --- a/src/storages/inMemory/InMemoryStorageCS.ts +++ b/src/storages/inMemory/InMemoryStorageCS.ts @@ -14,15 +14,19 @@ import { UniqueKeysCacheInMemoryCS } from './UniqueKeysCacheInMemoryCS'; * @param params parameters required by EventsCacheSync */ export function InMemoryStorageCSFactory(params: IStorageFactoryParams): IStorageSync { + const { settings: { scheduler: { impressionsQueueSize, eventsQueueSize, }, sync: { impressionsMode } } } = params; + + const splits = new SplitsCacheInMemory(); + const segments = new MySegmentsCacheInMemory(); return { - splits: new SplitsCacheInMemory(), - segments: new MySegmentsCacheInMemory(), - impressions: new ImpressionsCacheInMemory(params.impressionsQueueSize), - impressionCounts: params.impressionsMode !== DEBUG ? new ImpressionCountsCacheInMemory() : undefined, - events: new EventsCacheInMemory(params.eventsQueueSize), - telemetry: shouldRecordTelemetry(params) ? new TelemetryCacheInMemory() : undefined, - uniqueKeys: params.impressionsMode === NONE ? new UniqueKeysCacheInMemoryCS() : undefined, + splits, + segments, + impressions: new ImpressionsCacheInMemory(impressionsQueueSize), + impressionCounts: impressionsMode !== DEBUG ? new ImpressionCountsCacheInMemory() : undefined, + events: new EventsCacheInMemory(eventsQueueSize), + telemetry: shouldRecordTelemetry(params) ? new TelemetryCacheInMemory(splits, segments) : undefined, + uniqueKeys: impressionsMode === NONE ? new UniqueKeysCacheInMemoryCS() : undefined, // When using MEMORY we should clean all the caches to leave them empty destroy() { diff --git a/src/storages/inMemory/TelemetryCacheInMemory.ts b/src/storages/inMemory/TelemetryCacheInMemory.ts index 4d626e52..e62660e0 100644 --- a/src/storages/inMemory/TelemetryCacheInMemory.ts +++ b/src/storages/inMemory/TelemetryCacheInMemory.ts @@ -1,7 +1,7 @@ -import { ImpressionDataType, EventDataType, LastSync, HttpErrors, HttpLatencies, StreamingEvent, Method, OperationType, MethodExceptions, MethodLatencies } from '../../sync/submitters/types'; -import { LOCALHOST_MODE } from '../../utils/constants'; +import { ImpressionDataType, EventDataType, LastSync, HttpErrors, HttpLatencies, StreamingEvent, Method, OperationType, MethodExceptions, MethodLatencies, TelemetryUsageStatsPayload } from '../../sync/submitters/types'; +import { DEDUPED, DROPPED, LOCALHOST_MODE, QUEUED } from '../../utils/constants'; import { findLatencyIndex } from '../findLatencyIndex'; -import { IStorageFactoryParams, ITelemetryCacheSync } from '../types'; +import { ISegmentsCacheSync, ISplitsCacheSync, IStorageFactoryParams, ITelemetryCacheSync } from '../types'; const MAX_STREAMING_EVENTS = 20; const MAX_TAGS = 10; @@ -19,12 +19,48 @@ const ACCEPTANCE_RANGE = 0.001; * Record telemetry if mode is not localhost. * All factory instances track telemetry on server-side, and 0.1% on client-side. */ -export function shouldRecordTelemetry(params: IStorageFactoryParams) { - return params.mode !== LOCALHOST_MODE && (params.matchingKey === undefined || Math.random() <= ACCEPTANCE_RANGE); +export function shouldRecordTelemetry({ settings }: IStorageFactoryParams) { + return settings.mode !== LOCALHOST_MODE && (settings.core.key === undefined || Math.random() <= ACCEPTANCE_RANGE); } export class TelemetryCacheInMemory implements ITelemetryCacheSync { + constructor(private splits?: ISplitsCacheSync, private segments?: ISegmentsCacheSync) { } + + // isEmpty flag + private e = true; + + isEmpty() { return this.e; } + + clear() { /* no-op */ } + + pop(): TelemetryUsageStatsPayload { + this.e = true; + + return { + lS: this.getLastSynchronization(), + mL: this.popLatencies(), + mE: this.popExceptions(), + hE: this.popHttpErrors(), + hL: this.popHttpLatencies(), + tR: this.popTokenRefreshes(), + aR: this.popAuthRejections(), + iQ: this.getImpressionStats(QUEUED), + iDe: this.getImpressionStats(DEDUPED), + iDr: this.getImpressionStats(DROPPED), + spC: this.splits && this.splits.getSplitNames().length, + seC: this.segments && this.segments.getRegisteredSegments().length, + skC: this.segments && this.segments.getKeysCount(), + sL: this.getSessionLength(), + eQ: this.getEventStats(QUEUED), + eD: this.getEventStats(DROPPED), + sE: this.popStreamingEvents(), + t: this.popTags(), + }; + } + + /** Config stats */ + private timeUntilReady?: number; getTimeUntilReady() { @@ -55,6 +91,8 @@ export class TelemetryCacheInMemory implements ITelemetryCacheSync { this.notReadyUsage++; } + /** Usage stats */ + private impressionStats = [0, 0, 0]; getImpressionStats(type: ImpressionDataType) { @@ -63,6 +101,7 @@ export class TelemetryCacheInMemory implements ITelemetryCacheSync { recordImpressionStats(type: ImpressionDataType, count: number) { this.impressionStats[type] += count; + this.e = false; } private eventStats = [0, 0]; @@ -73,9 +112,9 @@ export class TelemetryCacheInMemory implements ITelemetryCacheSync { recordEventStats(type: EventDataType, count: number) { this.eventStats[type] += count; + this.e = false; } - // @ts-expect-error private lastSync: LastSync = {}; getLastSynchronization() { @@ -84,40 +123,35 @@ export class TelemetryCacheInMemory implements ITelemetryCacheSync { recordSuccessfulSync(resource: OperationType, timeMs: number) { this.lastSync[resource] = timeMs; + this.e = false; } - // @ts-expect-error private httpErrors: HttpErrors = {}; popHttpErrors() { - const result = this.httpErrors; // @ts-expect-error + const result = this.httpErrors; this.httpErrors = {}; return result; } recordHttpError(resource: OperationType, status: number) { - if (!this.httpErrors[resource]) this.httpErrors[resource] = {}; - if (!this.httpErrors[resource][status]) { - this.httpErrors[resource][status] = 1; - } else { - this.httpErrors[resource][status]++; - } + const statusErrors = (this.httpErrors[resource] = this.httpErrors[resource] || {}); + statusErrors[status] = (statusErrors[status] || 0) + 1; + this.e = false; } - // @ts-expect-error private httpLatencies: HttpLatencies = {}; popHttpLatencies() { - const result = this.httpLatencies; // @ts-expect-error + const result = this.httpLatencies; this.httpLatencies = {}; return result; } recordHttpLatency(resource: OperationType, latencyMs: number) { - if (!this.httpLatencies[resource]) { - this.httpLatencies[resource] = newBuckets(); - } - this.httpLatencies[resource][findLatencyIndex(latencyMs)]++; + const latencyBuckets = (this.httpLatencies[resource] = this.httpLatencies[resource] || newBuckets()); + latencyBuckets[findLatencyIndex(latencyMs)]++; + this.e = false; } private authRejections = 0; @@ -130,6 +164,7 @@ export class TelemetryCacheInMemory implements ITelemetryCacheSync { recordAuthRejections() { this.authRejections++; + this.e = false; } private tokenRefreshes = 0; @@ -142,6 +177,7 @@ export class TelemetryCacheInMemory implements ITelemetryCacheSync { recordTokenRefreshes() { this.tokenRefreshes++; + this.e = false; } private streamingEvents: StreamingEvent[] = [] @@ -154,6 +190,7 @@ export class TelemetryCacheInMemory implements ITelemetryCacheSync { if (this.streamingEvents.length < MAX_STREAMING_EVENTS) { this.streamingEvents.push(streamingEvent); } + this.e = false; } private tags: string[] = []; @@ -166,6 +203,7 @@ export class TelemetryCacheInMemory implements ITelemetryCacheSync { if (this.tags.length < MAX_TAGS) { this.tags.push(tag); } + this.e = false; } private sessionLength?: number; @@ -176,39 +214,34 @@ export class TelemetryCacheInMemory implements ITelemetryCacheSync { recordSessionLength(ms: number) { this.sessionLength = ms; + this.e = false; } - // @ts-expect-error private exceptions: MethodExceptions = {}; popExceptions() { - const result = this.exceptions; // @ts-expect-error + const result = this.exceptions; this.exceptions = {}; return result; } recordException(method: Method) { - if (!this.exceptions[method]) { - this.exceptions[method] = 1; - } else { - this.exceptions[method]++; - } + this.exceptions[method] = (this.exceptions[method] || 0) + 1; + this.e = false; } - // @ts-expect-error private latencies: MethodLatencies = {}; popLatencies() { - const result = this.latencies; // @ts-expect-error + const result = this.latencies; this.latencies = {}; return result; } recordLatency(method: Method, latencyMs: number) { - if (!this.latencies[method]) { - this.latencies[method] = newBuckets(); - } - this.latencies[method][findLatencyIndex(latencyMs)]++; + const latencyBuckets = (this.latencies[method] = this.latencies[method] || newBuckets()); + latencyBuckets[findLatencyIndex(latencyMs)]++; + this.e = false; } } diff --git a/src/storages/inMemory/UniqueKeysCacheInMemory.ts b/src/storages/inMemory/UniqueKeysCacheInMemory.ts index 675dabe4..ad386ff2 100644 --- a/src/storages/inMemory/UniqueKeysCacheInMemory.ts +++ b/src/storages/inMemory/UniqueKeysCacheInMemory.ts @@ -3,12 +3,31 @@ import { ISet, setToArray, _Set } from '../../utils/lang/sets'; import { UniqueKeysPayloadSs } from '../../sync/submitters/types'; import { DEFAULT_CACHE_SIZE } from '../inRedis/constants'; +/** + * Converts `uniqueKeys` data from cache into request payload for SS. + */ +export function fromUniqueKeysCollector(uniqueKeys: { [featureName: string]: ISet }): UniqueKeysPayloadSs { + const payload = []; + const featureNames = Object.keys(uniqueKeys); + for (let i = 0; i < featureNames.length; i++) { + const featureName = featureNames[i]; + const userKeys = setToArray(uniqueKeys[featureName]); + const uniqueKeysPayload = { + f: featureName, + ks: userKeys + }; + + payload.push(uniqueKeysPayload); + } + return { keys: payload }; +} + export class UniqueKeysCacheInMemory implements IUniqueKeysCacheBase { protected onFullQueue?: () => void; private readonly maxStorage: number; private uniqueTrackerSize = 0; - protected uniqueKeysTracker: { [keys: string]: ISet }; + protected uniqueKeysTracker: { [featureName: string]: ISet }; constructor(uniqueKeysQueueSize = DEFAULT_CACHE_SIZE) { this.maxStorage = uniqueKeysQueueSize; @@ -20,15 +39,13 @@ export class UniqueKeysCacheInMemory implements IUniqueKeysCacheBase { } /** - * Store unique keys in sequential order - * key: string = feature name. - * value: Set = set of unique keys. + * Store unique keys per feature. */ - track(key: string, featureName: string) { + track(userKey: string, featureName: string) { if (!this.uniqueKeysTracker[featureName]) this.uniqueKeysTracker[featureName] = new _Set(); const tracker = this.uniqueKeysTracker[featureName]; - if (!tracker.has(key)) { - tracker.add(key); + if (!tracker.has(userKey)) { + tracker.add(userKey); this.uniqueTrackerSize++; } if (this.uniqueTrackerSize >= this.maxStorage && this.onFullQueue) { @@ -50,7 +67,7 @@ export class UniqueKeysCacheInMemory implements IUniqueKeysCacheBase { pop() { const data = this.uniqueKeysTracker; this.uniqueKeysTracker = {}; - return this.fromUniqueKeysCollector(data); + return fromUniqueKeysCollector(data); } /** @@ -60,23 +77,4 @@ export class UniqueKeysCacheInMemory implements IUniqueKeysCacheBase { return Object.keys(this.uniqueKeysTracker).length === 0; } - /** - * Converts `uniqueKeys` data from cache into request payload for SS. - */ - private fromUniqueKeysCollector(uniqueKeys: { [featureName: string]: ISet }): UniqueKeysPayloadSs { - const payload = []; - const featureNames = Object.keys(uniqueKeys); - for (let i = 0; i < featureNames.length; i++) { - const featureName = featureNames[i]; - const featureKeys = setToArray(uniqueKeys[featureName]); - const uniqueKeysPayload = { - f: featureName, - ks: featureKeys - }; - - payload.push(uniqueKeysPayload); - } - return { keys: payload }; - } - } diff --git a/src/storages/inMemory/UniqueKeysCacheInMemoryCS.ts b/src/storages/inMemory/UniqueKeysCacheInMemoryCS.ts index 350c0cce..b8748063 100644 --- a/src/storages/inMemory/UniqueKeysCacheInMemoryCS.ts +++ b/src/storages/inMemory/UniqueKeysCacheInMemoryCS.ts @@ -8,7 +8,7 @@ export class UniqueKeysCacheInMemoryCS implements IUniqueKeysCacheBase { private onFullQueue?: () => void; private readonly maxStorage: number; private uniqueTrackerSize = 0; - private uniqueKeysTracker: { [keys: string]: ISet }; + private uniqueKeysTracker: { [userKey: string]: ISet }; /** * @@ -25,14 +25,12 @@ export class UniqueKeysCacheInMemoryCS implements IUniqueKeysCacheBase { } /** - * Store unique keys in sequential order - * key: string = key. - * value: HashSet = set of split names. + * Store unique keys per feature. */ - track(key: string, featureName: string) { + track(userKey: string, featureName: string) { - if (!this.uniqueKeysTracker[key]) this.uniqueKeysTracker[key] = new _Set(); - const tracker = this.uniqueKeysTracker[key]; + if (!this.uniqueKeysTracker[userKey]) this.uniqueKeysTracker[userKey] = new _Set(); + const tracker = this.uniqueKeysTracker[userKey]; if (!tracker.has(featureName)) { tracker.add(featureName); this.uniqueTrackerSize++; @@ -69,14 +67,14 @@ export class UniqueKeysCacheInMemoryCS implements IUniqueKeysCacheBase { /** * Converts `uniqueKeys` data from cache into request payload. */ - private fromUniqueKeysCollector(uniqueKeys: { [featureName: string]: ISet }): UniqueKeysPayloadCs { + private fromUniqueKeysCollector(uniqueKeys: { [userKey: string]: ISet }): UniqueKeysPayloadCs { const payload = []; - const featureKeys = Object.keys(uniqueKeys); - for (let k = 0; k < featureKeys.length; k++) { - const featureKey = featureKeys[k]; - const featureNames = setToArray(uniqueKeys[featureKey]); + const userKeys = Object.keys(uniqueKeys); + for (let k = 0; k < userKeys.length; k++) { + const userKey = userKeys[k]; + const featureNames = setToArray(uniqueKeys[userKey]); const uniqueKeysPayload = { - k: featureKey, + k: userKey, fs: featureNames }; diff --git a/src/storages/inMemory/__tests__/TelemetryCacheInMemory.spec.ts b/src/storages/inMemory/__tests__/TelemetryCacheInMemory.spec.ts index a3d6d854..85cc43fe 100644 --- a/src/storages/inMemory/__tests__/TelemetryCacheInMemory.spec.ts +++ b/src/storages/inMemory/__tests__/TelemetryCacheInMemory.spec.ts @@ -179,7 +179,6 @@ describe('TELEMETRY CACHE', () => { expect(cache.popTags()).toEqual([]); }); - test('method exceptions', () => { expect(cache.popExceptions()).toEqual({}); methods.forEach((method) => { @@ -209,4 +208,23 @@ describe('TELEMETRY CACHE', () => { expect(cache.popLatencies()).toEqual({}); }); + test('"isEmpty" and "pop" methods', () => { + const cache = new TelemetryCacheInMemory(); + const expectedEmptyPayload = { + lS: {}, mL: {}, mE: {}, hE: {}, hL: {}, tR: 0, aR: 0, iQ: 0, iDe: 0, iDr: 0, spC: undefined, seC: undefined, skC: undefined, eQ: 0, eD: 0, sE: [], t: [] + }; + + // Initially, the cache is empty + expect(cache.isEmpty()).toBe(true); + expect(cache.pop()).toEqual(expectedEmptyPayload); + + // Record some data to flag the cache as not empty + cache.recordException(TRACK); + expect(cache.isEmpty()).toBe(false); + + // Pop the data and check that the cache is empty again + expect(cache.pop()).toEqual({ ...expectedEmptyPayload, mE: { 'tr': 1 } }); + expect(cache.isEmpty()).toBe(true); + }); + }); diff --git a/src/storages/inRedis/ImpressionCountsCacheInRedis.ts b/src/storages/inRedis/ImpressionCountsCacheInRedis.ts index 9a4a0a1f..b0c563c0 100644 --- a/src/storages/inRedis/ImpressionCountsCacheInRedis.ts +++ b/src/storages/inRedis/ImpressionCountsCacheInRedis.ts @@ -1,5 +1,7 @@ import { Redis } from 'ioredis'; import { ILogger } from '../../logger/types'; +import { ImpressionCountsPayload } from '../../sync/submitters/types'; +import { forOwn } from '../../utils/lang'; import { ImpressionCountsCacheInMemory } from '../inMemory/ImpressionCountsCacheInMemory'; import { LOG_PREFIX, REFRESH_RATE, TTL_REFRESH } from './constants'; @@ -20,7 +22,7 @@ export class ImpressionCountsCacheInRedis extends ImpressionCountsCacheInMemory this.onFullQueue = () => { this.postImpressionCountsInRedis(); }; } - private postImpressionCountsInRedis(){ + private postImpressionCountsInRedis() { const counts = this.pop(); const keys = Object.keys(counts); if (!keys.length) return Promise.resolve(false); @@ -50,4 +52,44 @@ export class ImpressionCountsCacheInRedis extends ImpressionCountsCacheInMemory clearInterval(this.intervalId); return this.postImpressionCountsInRedis(); } + + // Async consumer API, used by synchronizer + getImpressionsCount(): Promise { + return this.redis.hgetall(this.key) + .then(counts => { + if (!Object.keys(counts).length) return undefined; + + this.redis.del(this.key).catch(() => { /* no-op */ }); + + const pf: ImpressionCountsPayload['pf'] = []; + + forOwn(counts, (count, key) => { + const nameAndTime = key.split('::'); + if (nameAndTime.length !== 2) { + this.log.error(`${LOG_PREFIX}Error spliting key ${key}`); + return; + } + + const timeFrame = parseInt(nameAndTime[1]); + if (isNaN(timeFrame)) { + this.log.error(`${LOG_PREFIX}Error parsing time frame ${nameAndTime[1]}`); + return; + } + + const rawCount = parseInt(count); + if (isNaN(rawCount)) { + this.log.error(`${LOG_PREFIX}Error parsing raw count ${count}`); + return; + } + + pf.push({ + f: nameAndTime[0], + m: timeFrame, + rc: rawCount, + }); + }); + + return { pf }; + }); + } } diff --git a/src/storages/inRedis/ImpressionsCacheInRedis.ts b/src/storages/inRedis/ImpressionsCacheInRedis.ts index d4bd2d96..12ee277c 100644 --- a/src/storages/inRedis/ImpressionsCacheInRedis.ts +++ b/src/storages/inRedis/ImpressionsCacheInRedis.ts @@ -4,6 +4,7 @@ import { ImpressionDTO } from '../../types'; import { Redis } from 'ioredis'; import { StoredImpressionWithMetadata } from '../../sync/submitters/types'; import { ILogger } from '../../logger/types'; +import { impressionsToJSON } from '../utils'; const IMPRESSIONS_TTL_REFRESH = 3600; // 1 hr @@ -24,7 +25,7 @@ export class ImpressionsCacheInRedis implements IImpressionsCacheAsync { track(impressions: ImpressionDTO[]): Promise { // @ts-ignore return this.redis.rpush( this.key, - this._toJSON(impressions) + impressionsToJSON(impressions, this.metadata), ).then(queuedCount => { // If this is the creation of the key on Redis, set the expiration for it in 1hr. if (queuedCount === impressions.length) { @@ -33,27 +34,6 @@ export class ImpressionsCacheInRedis implements IImpressionsCacheAsync { }); } - private _toJSON(impressions: ImpressionDTO[]): string[] { - return impressions.map(impression => { - const { - keyName, bucketingKey, feature, treatment, label, time, changeNumber - } = impression; - - return JSON.stringify({ - m: this.metadata, - i: { - k: keyName, - b: bucketingKey, - f: feature, - t: treatment, - r: label, - c: changeNumber, - m: time - } - }); - }); - } - count(): Promise { return this.redis.llen(this.key).catch(() => 0); } diff --git a/src/storages/inRedis/TelemetryCacheInRedis.ts b/src/storages/inRedis/TelemetryCacheInRedis.ts index b866fbe2..ab50fbe5 100644 --- a/src/storages/inRedis/TelemetryCacheInRedis.ts +++ b/src/storages/inRedis/TelemetryCacheInRedis.ts @@ -1,6 +1,6 @@ import { ILogger } from '../../logger/types'; import { Method, MultiConfigs, MultiMethodExceptions, MultiMethodLatencies } from '../../sync/submitters/types'; -import { KeyBuilderSS, parseExceptionField, parseLatencyField, parseMetadata } from '../KeyBuilderSS'; +import { KeyBuilderSS } from '../KeyBuilderSS'; import { ITelemetryCacheAsync } from '../types'; import { findLatencyIndex } from '../findLatencyIndex'; import { Redis } from 'ioredis'; @@ -9,6 +9,7 @@ import { CONSUMER_MODE, STORAGE_REDIS } from '../../utils/constants'; import { isNaNNumber, isString } from '../../utils/lang'; import { _Map } from '../../utils/lang/maps'; import { MAX_LATENCY_BUCKET_COUNT, newBuckets } from '../inMemory/TelemetryCacheInMemory'; +import { parseLatencyField, parseExceptionField, parseMetadata } from '../utils'; export class TelemetryCacheInRedis implements ITelemetryCacheAsync { @@ -76,7 +77,7 @@ export class TelemetryCacheInRedis implements ITelemetryCacheAsync { tr: newBuckets(), }); - result.get(metadata)![method][bucket] = count; + result.get(metadata)![method]![bucket] = count; }); return this.redis.del(this.keys.latencyPrefix).then(() => result); diff --git a/src/storages/inRedis/UniqueKeysCacheInRedis.ts b/src/storages/inRedis/UniqueKeysCacheInRedis.ts index ba1d3053..bfd96b31 100644 --- a/src/storages/inRedis/UniqueKeysCacheInRedis.ts +++ b/src/storages/inRedis/UniqueKeysCacheInRedis.ts @@ -5,6 +5,7 @@ import { setToArray } from '../../utils/lang/sets'; import { DEFAULT_CACHE_SIZE, REFRESH_RATE, TTL_REFRESH } from './constants'; import { LOG_PREFIX } from './constants'; import { ILogger } from '../../logger/types'; +import { UniqueKeysItemSs } from '../../sync/submitters/types'; export class UniqueKeysCacheInRedis extends UniqueKeysCacheInMemory implements IUniqueKeysCacheBase { @@ -20,7 +21,7 @@ export class UniqueKeysCacheInRedis extends UniqueKeysCacheInMemory implements I this.key = key; this.redis = redis; this.refreshRate = refreshRate; - this.onFullQueue = () => {this.postUniqueKeysInRedis();}; + this.onFullQueue = () => { this.postUniqueKeysInRedis(); }; } private postUniqueKeysInRedis() { @@ -62,4 +63,15 @@ export class UniqueKeysCacheInRedis extends UniqueKeysCacheInMemory implements I return this.postUniqueKeysInRedis(); } + /** + * Async consumer API, used by synchronizer. + * @param count number of items to pop from the queue. If not provided or equal 0, all items will be popped. + */ + popNRaw(count = 0): Promise { + return this.redis.lrange(this.key, 0, count - 1).then(uniqueKeyItems => { + return this.redis.ltrim(this.key, uniqueKeyItems.length, -1) + .then(() => uniqueKeyItems.map(uniqueKeyItem => JSON.parse(uniqueKeyItem))); + }); + } + } diff --git a/src/storages/inRedis/__tests__/ImpressionCountsCacheInRedis.spec.ts b/src/storages/inRedis/__tests__/ImpressionCountsCacheInRedis.spec.ts index 169e00b2..e28011c6 100644 --- a/src/storages/inRedis/__tests__/ImpressionCountsCacheInRedis.spec.ts +++ b/src/storages/inRedis/__tests__/ImpressionCountsCacheInRedis.spec.ts @@ -145,46 +145,44 @@ describe('IMPRESSION COUNTS CACHE IN REDIS', () => { }); - test('Should call "onFullQueueCb" when the queue is full.', (done) => { + test('Should call "onFullQueueCb" when the queue is full. "getImpressionsCount" should pop data.', async () => { const connection = new Redis(); - const shortCounter = new ImpressionCountsCacheInRedis(loggerMock, key, connection, 5); + const counter = new ImpressionCountsCacheInRedis(loggerMock, key, connection, 5); // Clean up in case there are still keys there. - connection.del(key); - shortCounter.track('feature1', timestamp + 1, 1); - shortCounter.track('feature1', timestamp + 2, 1); - shortCounter.track('feature2', timestamp + 3, 2); - connection.hgetall(key, (err, data) => { - expect(data).toStrictEqual({}); - }); + await connection.del(key); + counter.track('feature1', timestamp + 1, 1); + counter.track('feature1', timestamp + 2, 1); + counter.track('feature2', timestamp + 3, 2); + + expect(await connection.hgetall(key)).toStrictEqual({}); - shortCounter.track('feature2', nextHourTimestamp, 1); + counter.track('feature2', nextHourTimestamp, 1); const expected1 = {}; expected1[`feature1::${truncateTimeFrame(timestamp)}`] = '2'; expected1[`feature2::${truncateTimeFrame(timestamp)}`] = '2'; expected1[`feature2::${truncateTimeFrame(nextHourTimestamp)}`] = '1'; - connection.hgetall(key, (err, data) => { - expect(data).toStrictEqual(expected1); - }); + expect(await connection.hgetall(key)).toStrictEqual(expected1); - const expected2 = {}; - expected2[`feature1::${truncateTimeFrame(timestamp)}`] = '4'; - expected2[`feature1::${truncateTimeFrame(nextHourTimestamp)}`] = '3'; - expected2[`feature2::${truncateTimeFrame(timestamp)}`] = '4'; - expected2[`feature2::${truncateTimeFrame(nextHourTimestamp)}`] = '1'; - - shortCounter.track('feature1', timestamp + 1, 1); - shortCounter.track('feature1', timestamp + 2, 1); - shortCounter.track('feature2', timestamp + 3, 2); - shortCounter.track('feature1', nextHourTimestamp + 2, 3); - - connection.hgetall(key, async (err, data) => { - expect(data).toStrictEqual(expected2); - await connection.del(key); - await connection.quit(); - done(); + counter.track('feature1', timestamp + 1, 1); + counter.track('feature1', timestamp + 2, 1); + counter.track('feature2', timestamp + 3, 2); + counter.track('feature1', nextHourTimestamp + 2, 3); + + // Validate `getImpressionsCount` method + expect(await counter.getImpressionsCount()).toStrictEqual({ // pop data + pf: [ + { f: 'feature1', m: truncateTimeFrame(timestamp), rc: 4 }, + { f: 'feature2', m: truncateTimeFrame(timestamp), rc: 4 }, + { f: 'feature2', m: truncateTimeFrame(nextHourTimestamp), rc: 1 }, + { f: 'feature1', m: truncateTimeFrame(nextHourTimestamp), rc: 3 }, + ] }); + expect(await counter.getImpressionsCount()).toStrictEqual(undefined); // try to pop data again + expect(await connection.hgetall(key)).toStrictEqual({}); + await connection.del(key); + await connection.quit(); }); }); diff --git a/src/storages/inRedis/__tests__/SplitsCacheInRedis.spec.ts b/src/storages/inRedis/__tests__/SplitsCacheInRedis.spec.ts index ca762e00..8cff408b 100644 --- a/src/storages/inRedis/__tests__/SplitsCacheInRedis.spec.ts +++ b/src/storages/inRedis/__tests__/SplitsCacheInRedis.spec.ts @@ -24,7 +24,8 @@ describe('SPLITS CACHE REDIS', () => { let values = await cache.getAll(); - expect(values).toEqual([splitWithAccountTT, splitWithUserTT]); + expect(values).toHaveLength(2); + expect(values).toEqual(values[0].name === 'lol1' ? [splitWithUserTT, splitWithAccountTT] : [splitWithAccountTT, splitWithUserTT]); let splitNames = await cache.getSplitNames(); diff --git a/src/storages/inRedis/__tests__/UniqueKeysCacheInRedis.spec.ts b/src/storages/inRedis/__tests__/UniqueKeysCacheInRedis.spec.ts index d44ec4d3..e64e82a5 100644 --- a/src/storages/inRedis/__tests__/UniqueKeysCacheInRedis.spec.ts +++ b/src/storages/inRedis/__tests__/UniqueKeysCacheInRedis.spec.ts @@ -6,10 +6,10 @@ import { loggerMock } from '../../../logger/__tests__/sdkLogger.mock'; import { RedisMock } from '../../../utils/redis/RedisMock'; describe('UNIQUE KEYS CACHE IN REDIS', () => { + const key = 'unique_key_post'; test('should incrementally store values, clear the queue, and tell if it is empty', async () => { const connection = new Redis(); - const key = 'unique_key_post'; const cache = new UniqueKeysCacheInRedis(loggerMock, key, connection); @@ -52,7 +52,6 @@ describe('UNIQUE KEYS CACHE IN REDIS', () => { test('Should call "onFullQueueCb" when the queue is full.', async () => { let cbCalled = 0; const connection = new Redis(); - const key = 'unique_key_post'; const cache = new UniqueKeysCacheInRedis(loggerMock, key, connection, 3); // small uniqueKeysCache size to be reached cache.setOnFullQueueCb(() => { cbCalled++; cache.clear(); }); @@ -81,9 +80,6 @@ describe('UNIQUE KEYS CACHE IN REDIS', () => { test('post unique keys in redis method', async () => { const connection = new Redis(); - const key = 'unique_key_post'; - // Clean up in case there are still keys there. - await connection.del(key); const cache = new UniqueKeysCacheInRedis(loggerMock, key, connection, 20); cache.track('key1', 'feature1'); @@ -105,8 +101,6 @@ describe('UNIQUE KEYS CACHE IN REDIS', () => { await connection.del(key); await connection.quit(); }); - - }); test('start and stop task', (done) => { @@ -144,12 +138,8 @@ describe('UNIQUE KEYS CACHE IN REDIS', () => { }, 2 * refreshRate + 30); }); - test('Should call "onFullQueueCb" when the queue is full.', async () => { + test('Should call "onFullQueueCb" when the queue is full. "popNRaw" should pop items.', async () => { const connection = new Redis(); - const key = 'unique_key_post'; - - // Clean up in case there are still keys there. - await connection.del(key); const cache = new UniqueKeysCacheInRedis(loggerMock, key, connection, 3); @@ -168,27 +158,26 @@ describe('UNIQUE KEYS CACHE IN REDIS', () => { JSON.stringify({ 'f': 'feature3', 'ks': ['key2'] }) ]; - connection.lrange(key, 0, 10, (err, data) => { - expect(data).toStrictEqual(expected1); - }); - - const expected2 = [ - ...expected1, - JSON.stringify({ 'f': 'feature4', 'ks': ['key2'] }), - JSON.stringify({ 'f': 'feature5', 'ks': ['key3'] }), - JSON.stringify({ 'f': 'feature6', 'ks': ['key2'] }) - ]; + let data = await connection.lrange(key, 0, 10); + expect(data).toStrictEqual(expected1); cache.track('key2', 'feature4'); cache.track('key2', 'feature4'); cache.track('key3', 'feature5'); cache.track('key2', 'feature6'); - connection.lrange(key, 0, 10, async (err, data) => { - expect(data).toStrictEqual(expected2); - await connection.del(key); - await connection.quit(); - }); + // Validate `popNRaw` method + let poped = await cache.popNRaw(2); // pop two items + expect(poped).toEqual([{ f: 'feature1', ks: ['key1'] }, { f: 'feature2', ks: ['key1'] }]); + poped = await cache.popNRaw(); // pop remaining items + expect(poped).toEqual([{ f: 'feature3', ks: ['key2'] }, { f: 'feature4', ks: ['key2'] }, { f: 'feature5', ks: ['key3'] }, { f: 'feature6', ks: ['key2'] }]); + poped = await cache.popNRaw(100); // try to pop more items when the queue is empty + expect(poped).toEqual([]); + + data = await connection.lrange(key, 0, 10); + expect(data).toStrictEqual([]); + await connection.del(key); + await connection.quit(); }); }); diff --git a/src/storages/inRedis/index.ts b/src/storages/inRedis/index.ts index 3bcd7549..0b6f923f 100644 --- a/src/storages/inRedis/index.ts +++ b/src/storages/inRedis/index.ts @@ -10,6 +10,7 @@ import { DEBUG, NONE, STORAGE_REDIS } from '../../utils/constants'; import { TelemetryCacheInRedis } from './TelemetryCacheInRedis'; import { UniqueKeysCacheInRedis } from './UniqueKeysCacheInRedis'; import { ImpressionCountsCacheInRedis } from './ImpressionCountsCacheInRedis'; +import { metadataBuilder } from '../utils'; export interface InRedisStorageOptions { prefix?: string @@ -24,7 +25,9 @@ export function InRedisStorage(options: InRedisStorageOptions = {}): IStorageAsy const prefix = validatePrefix(options.prefix); - function InRedisStorageFactory({ log, metadata, onReadyCb, impressionsMode }: IStorageFactoryParams): IStorageAsync { + function InRedisStorageFactory(params: IStorageFactoryParams): IStorageAsync { + const { onReadyCb, settings, settings: { log, sync: { impressionsMode } } } = params; + const metadata = metadataBuilder(settings); const keys = new KeyBuilderSS(prefix, metadata); const redisClient = new RedisAdapter(log, options.options || {}); const telemetry = new TelemetryCacheInRedis(log, keys, redisClient); @@ -52,7 +55,7 @@ export function InRedisStorage(options: InRedisStorageOptions = {}): IStorageAsy // When using REDIS we should: // 1- Disconnect from the storage - destroy(): Promise{ + destroy(): Promise { let promises = []; if (impressionCountsCache) promises.push(impressionCountsCache.stop()); if (uniqueKeysCache) promises.push(uniqueKeysCache.stop()); diff --git a/src/storages/metadataBuilder.ts b/src/storages/metadataBuilder.ts deleted file mode 100644 index 2cfd9d2d..00000000 --- a/src/storages/metadataBuilder.ts +++ /dev/null @@ -1,11 +0,0 @@ -import { IMetadata } from '../dtos/types'; -import { ISettings } from '../types'; -import { UNKNOWN } from '../utils/constants'; - -export function metadataBuilder(settings: Pick): IMetadata { - return { - s: settings.version, - i: settings.runtime.ip || UNKNOWN, - n: settings.runtime.hostname || UNKNOWN, - }; -} diff --git a/src/storages/pluggable/ImpressionCountsCachePluggable.ts b/src/storages/pluggable/ImpressionCountsCachePluggable.ts index 725e8399..ad07af73 100644 --- a/src/storages/pluggable/ImpressionCountsCachePluggable.ts +++ b/src/storages/pluggable/ImpressionCountsCachePluggable.ts @@ -1,4 +1,5 @@ import { ILogger } from '../../logger/types'; +import { ImpressionCountsPayload } from '../../sync/submitters/types'; import { ImpressionCountsCacheInMemory } from '../inMemory/ImpressionCountsCacheInMemory'; import { REFRESH_RATE } from '../inRedis/constants'; import { IPluggableStorageWrapper } from '../types'; @@ -44,4 +45,48 @@ export class ImpressionCountsCachePluggable extends ImpressionCountsCacheInMemor clearInterval(this.intervalId); return this.storeImpressionCounts(); } + + // Async consumer API, used by synchronizer + getImpressionsCount(): Promise { + return this.wrapper.getKeysByPrefix(this.key) + .then(keys => { + return keys.length ? Promise.all(keys.map(key => this.wrapper.get(key))) + .then(counts => { + keys.forEach(key => this.wrapper.del(key).catch(() => { /* noop */ })); + + const pf = []; + + for (let i = 0; i < keys.length; i++) { + const key = keys[i]; + const count = counts[i]; + + const keyFeatureNameAndTime = key.split('::'); + if (keyFeatureNameAndTime.length !== 3) { + this.log.error(`${LOG_PREFIX}Error spliting key ${key}`); + continue; + } + + const timeFrame = parseInt(keyFeatureNameAndTime[2]); + if (isNaN(timeFrame)) { + this.log.error(`${LOG_PREFIX}Error parsing time frame ${keyFeatureNameAndTime[2]}`); + continue; + } + // @ts-ignore + const rawCount = parseInt(count); + if (isNaN(rawCount)) { + this.log.error(`${LOG_PREFIX}Error parsing raw count ${count}`); + continue; + } + + pf.push({ + f: keyFeatureNameAndTime[1], + m: timeFrame, + rc: rawCount, + }); + } + + return { pf }; + }) : undefined; + }); + } } diff --git a/src/storages/pluggable/ImpressionsCachePluggable.ts b/src/storages/pluggable/ImpressionsCachePluggable.ts index 4de3e33e..dede350d 100644 --- a/src/storages/pluggable/ImpressionsCachePluggable.ts +++ b/src/storages/pluggable/ImpressionsCachePluggable.ts @@ -1,8 +1,9 @@ import { IPluggableStorageWrapper, IImpressionsCacheAsync } from '../types'; import { IMetadata } from '../../dtos/types'; import { ImpressionDTO } from '../../types'; -import { ILogger } from '../../logger/types'; import { StoredImpressionWithMetadata } from '../../sync/submitters/types'; +import { ILogger } from '../../logger/types'; +import { impressionsToJSON } from '../utils'; export class ImpressionsCachePluggable implements IImpressionsCacheAsync { @@ -27,31 +28,10 @@ export class ImpressionsCachePluggable implements IImpressionsCacheAsync { track(impressions: ImpressionDTO[]): Promise { return this.wrapper.pushItems( this.key, - this._toJSON(impressions) + impressionsToJSON(impressions, this.metadata) ); } - private _toJSON(impressions: ImpressionDTO[]): string[] { - return impressions.map(impression => { - const { - keyName, bucketingKey, feature, treatment, label, time, changeNumber - } = impression; - - return JSON.stringify({ - m: this.metadata, - i: { - k: keyName, - b: bucketingKey, - f: feature, - t: treatment, - r: label, - c: changeNumber, - m: time - } - } as StoredImpressionWithMetadata); - }); - } - /** * Returns a promise that resolves with the count of stored impressions, or 0 if there was some error. * The promise will never be rejected. diff --git a/src/storages/pluggable/TelemetryCachePluggable.ts b/src/storages/pluggable/TelemetryCachePluggable.ts index ae593952..8f54089f 100644 --- a/src/storages/pluggable/TelemetryCachePluggable.ts +++ b/src/storages/pluggable/TelemetryCachePluggable.ts @@ -1,6 +1,6 @@ import { ILogger } from '../../logger/types'; import { Method, MultiConfigs, MultiMethodExceptions, MultiMethodLatencies } from '../../sync/submitters/types'; -import { KeyBuilderSS, parseExceptionField, parseLatencyField, parseMetadata } from '../KeyBuilderSS'; +import { KeyBuilderSS } from '../KeyBuilderSS'; import { IPluggableStorageWrapper, ITelemetryCacheAsync } from '../types'; import { findLatencyIndex } from '../findLatencyIndex'; import { getTelemetryConfigStats } from '../../sync/submitters/telemetrySubmitter'; @@ -8,6 +8,7 @@ import { CONSUMER_MODE, STORAGE_PLUGGABLE } from '../../utils/constants'; import { isString, isNaNNumber } from '../../utils/lang'; import { _Map } from '../../utils/lang/maps'; import { MAX_LATENCY_BUCKET_COUNT, newBuckets } from '../inMemory/TelemetryCacheInMemory'; +import { parseLatencyField, parseExceptionField, parseMetadata } from '../utils'; export class TelemetryCachePluggable implements ITelemetryCacheAsync { @@ -75,7 +76,7 @@ export class TelemetryCachePluggable implements ITelemetryCacheAsync { tr: newBuckets(), }); - result.get(metadata)![method][bucket] = count; + result.get(metadata)![method]![bucket] = count; } return Promise.all(latencyKeys.map((latencyKey) => this.wrapper.del(latencyKey))).then(() => result); diff --git a/src/storages/pluggable/UniqueKeysCachePluggable.ts b/src/storages/pluggable/UniqueKeysCachePluggable.ts index a6e5173b..01a4080c 100644 --- a/src/storages/pluggable/UniqueKeysCachePluggable.ts +++ b/src/storages/pluggable/UniqueKeysCachePluggable.ts @@ -4,6 +4,7 @@ import { setToArray } from '../../utils/lang/sets'; import { DEFAULT_CACHE_SIZE, REFRESH_RATE } from '../inRedis/constants'; import { LOG_PREFIX } from './constants'; import { ILogger } from '../../logger/types'; +import { UniqueKeysItemSs } from '../../sync/submitters/types'; export class UniqueKeysCachePluggable extends UniqueKeysCacheInMemory implements IUniqueKeysCacheBase { @@ -53,4 +54,14 @@ export class UniqueKeysCachePluggable extends UniqueKeysCacheInMemory implements return this.storeUniqueKeys(); } + /** + * Async consumer API, used by synchronizer. + * @param count number of items to pop from the queue. If not provided or equal 0, all items will be popped. + */ + popNRaw(count = 0): Promise { + return Promise.resolve(count || this.wrapper.getItemsCount(this.key)) + .then(count => this.wrapper.popItems(this.key, count)) + .then((uniqueKeyItems) => uniqueKeyItems.map(uniqueKeyItem => JSON.parse(uniqueKeyItem))); + } + } diff --git a/src/storages/pluggable/__tests__/ImpressionCountsCachePluggable.spec.ts b/src/storages/pluggable/__tests__/ImpressionCountsCachePluggable.spec.ts index 2075761b..0d83d664 100644 --- a/src/storages/pluggable/__tests__/ImpressionCountsCachePluggable.spec.ts +++ b/src/storages/pluggable/__tests__/ImpressionCountsCachePluggable.spec.ts @@ -61,7 +61,8 @@ describe('IMPRESSION COUNTS CACHE PLUGGABLE', () => { }); - test('Should call "onFullQueueCb" when the queue is full.', async () => { + test('Should call "onFullQueueCb" when the queue is full. "getImpressionsCount" should pop data.', async () => { + const key = 'other_key'; const counter = new ImpressionCountsCachePluggable(loggerMock, key, wrapperMock, 5); counter.track('feature1', timestamp + 1, 1); @@ -83,5 +84,15 @@ describe('IMPRESSION COUNTS CACHE PLUGGABLE', () => { ]); expect(counter.isEmpty()).toBe(true); + + // Validate `getImpressionsCount` method + expect(await counter.getImpressionsCount()).toStrictEqual({ // pop data + pf: [ + { f: 'feature1', m: truncateTimeFrame(timestamp), rc: 2 }, + { f: 'feature2', m: truncateTimeFrame(timestamp), rc: 2 }, + { f: 'feature2', m: truncateTimeFrame(nextHourTimestamp), rc: 1 }, + ] + }); + expect(await counter.getImpressionsCount()).toStrictEqual(undefined); // try to pop data again }); }); diff --git a/src/storages/pluggable/__tests__/TelemetryCachePluggable.spec.ts b/src/storages/pluggable/__tests__/TelemetryCachePluggable.spec.ts index c0e03563..30062718 100644 --- a/src/storages/pluggable/__tests__/TelemetryCachePluggable.spec.ts +++ b/src/storages/pluggable/__tests__/TelemetryCachePluggable.spec.ts @@ -43,7 +43,7 @@ test('TELEMETRY CACHE PLUGGABLE', async () => { const latencies = await cache.popLatencies(); latencies.forEach((latency, m) => { expect(JSON.parse(m)).toEqual(metadata); - expect(latency.tr[2]).toBe(2); + expect(latency.tr![2]).toBe(2); }); // popExceptions diff --git a/src/storages/pluggable/__tests__/UniqueKeysCachePluggable.spec.ts b/src/storages/pluggable/__tests__/UniqueKeysCachePluggable.spec.ts index 16ff46eb..81220780 100644 --- a/src/storages/pluggable/__tests__/UniqueKeysCachePluggable.spec.ts +++ b/src/storages/pluggable/__tests__/UniqueKeysCachePluggable.spec.ts @@ -49,7 +49,8 @@ describe('UNIQUE KEYS CACHE PLUGGABLE', () => { }, 2 * refreshRate + 20); }); - test('Should call "onFullQueueCb" when the queue is full.', async () => { + test('Should call "onFullQueueCb" when the queue is full. "popNRaw" should pop items.', async () => { + const key = 'other_key'; const cache = new UniqueKeysCachePluggable(loggerMock, key, wrapperMock, 3); cache.track('key1', 'feature1'); @@ -66,11 +67,19 @@ describe('UNIQUE KEYS CACHE PLUGGABLE', () => { await new Promise(resolve => setTimeout(resolve)); expect(wrapperMock.pushItems.mock.calls).toEqual([ - [key, [JSON.stringify({ 'f': 'feature1', 'ks': ['key1'] })]], - [key, [JSON.stringify({ 'f': 'feature2', 'ks': ['key1'] })]], - [key, [JSON.stringify({ 'f': 'feature3', 'ks': ['key2'] })]] + [key, [JSON.stringify({ f: 'feature1', ks: ['key1'] })]], + [key, [JSON.stringify({ f: 'feature2', ks: ['key1'] })]], + [key, [JSON.stringify({ f: 'feature3', ks: ['key2'] })]] ]); expect(cache.isEmpty()).toBe(true); + + // Validate `popNRaw` method + let poped = await cache.popNRaw(2); // pop two items + expect(poped).toEqual([{ f: 'feature1', ks: ['key1'] }, { f: 'feature2', ks: ['key1'] }]); + poped = await cache.popNRaw(); // pop remaining items + expect(poped).toEqual([{ f: 'feature3', ks: ['key2'] }]); + poped = await cache.popNRaw(100); // try to pop more items when the queue is empty + expect(poped).toEqual([]); }); }); diff --git a/src/storages/pluggable/__tests__/index.spec.ts b/src/storages/pluggable/__tests__/index.spec.ts index 301bd8ba..759bd9a8 100644 --- a/src/storages/pluggable/__tests__/index.spec.ts +++ b/src/storages/pluggable/__tests__/index.spec.ts @@ -1,11 +1,11 @@ // @ts-nocheck // Mocks -import { loggerMock } from '../../../logger/__tests__/sdkLogger.mock'; +import { fullSettings } from '../../../utils/settingsValidation/__tests__/settings.mocks'; import { IStorageFactoryParams } from '../../types'; import { wrapperMock, wrapperMockFactory } from './wrapper.mock'; -const metadata = { s: 'version', i: 'ip', n: 'hostname' }; +// const metadata = { s: 'version', i: 'ip', n: 'hostname' }; const prefix = 'some_prefix'; // Test target @@ -16,8 +16,7 @@ import { CONSUMER_PARTIAL_MODE } from '../../../utils/constants'; describe('PLUGGABLE STORAGE', () => { const internalSdkParams: IStorageFactoryParams = { - log: loggerMock, - metadata, + settings: fullSettings, onReadyCb: jest.fn() }; @@ -73,7 +72,7 @@ describe('PLUGGABLE STORAGE', () => { test('creates a storage instance for partial consumer mode (events and impressions cache in memory)', async () => { const storageFactory = PluggableStorage({ prefix, wrapper: wrapperMock }); - const storage = storageFactory({ ...internalSdkParams, mode: CONSUMER_PARTIAL_MODE, optimize: true }); + const storage = storageFactory({ ...internalSdkParams, settings: { ...internalSdkParams.settings, mode: CONSUMER_PARTIAL_MODE } }); assertStorageInterface(storage); expect(wrapperMock.connect).toBeCalledTimes(1); @@ -91,4 +90,16 @@ describe('PLUGGABLE STORAGE', () => { const impResult = storage.impressions.track(['some data']); expect(impResult).toBe(undefined); }); + + test('creates a storage instance for the synchronizer', async () => { + const storageFactory = PluggableStorage({ prefix, wrapper: wrapperMock }); + const storage = storageFactory({ ...internalSdkParams, settings: { ...internalSdkParams.settings, mode: undefined } }); + + assertStorageInterface(storage); + + // All caches are present + expect(storage.telemetry).toBeDefined(); + expect(storage.impressionCounts).toBeDefined(); + expect(storage.uniqueKeys).toBeDefined(); + }); }); diff --git a/src/storages/pluggable/inMemoryWrapper.ts b/src/storages/pluggable/inMemoryWrapper.ts index cb3a77e8..c87c9a47 100644 --- a/src/storages/pluggable/inMemoryWrapper.ts +++ b/src/storages/pluggable/inMemoryWrapper.ts @@ -39,25 +39,25 @@ export function inMemoryWrapperFactory(connDelay?: number): IPluggableStorageWra getKeysByPrefix(prefix: string) { return Promise.resolve(Object.keys(_cache).filter(key => startsWith(key, prefix))); }, - incr(key: string) { + incr(key: string, increment = 1) { if (key in _cache) { - const count = toNumber(_cache[key]) + 1; + const count = toNumber(_cache[key]) + increment; if (isNaN(count)) return Promise.reject('Given key is not a number'); _cache[key] = count + ''; return Promise.resolve(count); } else { - _cache[key] = '1'; + _cache[key] = '' + increment; return Promise.resolve(1); } }, - decr(key: string) { + decr(key: string, decrement = 1) { if (key in _cache) { - const count = toNumber(_cache[key]) - 1; + const count = toNumber(_cache[key]) - decrement; if (isNaN(count)) return Promise.reject('Given key is not a number'); _cache[key] = count + ''; return Promise.resolve(count); } else { - _cache[key] = '-1'; + _cache[key] = '-' + decrement; return Promise.resolve(-1); } }, diff --git a/src/storages/pluggable/index.ts b/src/storages/pluggable/index.ts index bcefe9f1..b5af77f4 100644 --- a/src/storages/pluggable/index.ts +++ b/src/storages/pluggable/index.ts @@ -18,6 +18,7 @@ import { ImpressionCountsCachePluggable } from './ImpressionCountsCachePluggable import { UniqueKeysCachePluggable } from './UniqueKeysCachePluggable'; import { UniqueKeysCacheInMemory } from '../inMemory/UniqueKeysCacheInMemory'; import { UniqueKeysCacheInMemoryCS } from '../inMemory/UniqueKeysCacheInMemoryCS'; +import { metadataBuilder } from '../utils'; const NO_VALID_WRAPPER = 'Expecting pluggable storage `wrapper` in options, but no valid wrapper instance was provided.'; const NO_VALID_WRAPPER_INTERFACE = 'The provided wrapper instance doesn’t follow the expected interface. Check our docs.'; @@ -61,19 +62,29 @@ export function PluggableStorage(options: PluggableStorageOptions): IStorageAsyn const prefix = validatePrefix(options.prefix); function PluggableStorageFactory(params: IStorageFactoryParams): IStorageAsync { - const { log, metadata, onReadyCb, mode, eventsQueueSize, impressionsQueueSize, impressionsMode, matchingKey } = params; + const { onReadyCb, settings, settings: { log, mode, sync: { impressionsMode }, scheduler: { impressionsQueueSize, eventsQueueSize } } } = params; + const metadata = metadataBuilder(settings); const keys = new KeyBuilderSS(prefix, metadata); const wrapper = wrapperAdapter(log, options.wrapper); + + const isSyncronizer = mode === undefined; // If mode is not defined, the synchronizer is running const isPartialConsumer = mode === CONSUMER_PARTIAL_MODE; - const telemetry = shouldRecordTelemetry(params) ? - isPartialConsumer ? new TelemetryCacheInMemory() : new TelemetryCachePluggable(log, keys, wrapper) : + + const telemetry = shouldRecordTelemetry(params) || isSyncronizer ? + isPartialConsumer ? + new TelemetryCacheInMemory() : + new TelemetryCachePluggable(log, keys, wrapper) : undefined; - const impressionCountsCache = impressionsMode !== DEBUG ? - isPartialConsumer ? new ImpressionCountsCacheInMemory() : new ImpressionCountsCachePluggable(log, keys.buildImpressionsCountKey(), wrapper) : + + const impressionCountsCache = impressionsMode !== DEBUG || isSyncronizer ? + isPartialConsumer ? + new ImpressionCountsCacheInMemory() : + new ImpressionCountsCachePluggable(log, keys.buildImpressionsCountKey(), wrapper) : undefined; - const uniqueKeysCache = impressionsMode === NONE ? + + const uniqueKeysCache = impressionsMode === NONE || isSyncronizer ? isPartialConsumer ? - matchingKey === undefined ? new UniqueKeysCacheInMemory() : new UniqueKeysCacheInMemoryCS() : + settings.core.key === undefined ? new UniqueKeysCacheInMemory() : new UniqueKeysCacheInMemoryCS() : new UniqueKeysCachePluggable(log, keys.buildUniqueKeysKey(), wrapper) : undefined; @@ -81,12 +92,12 @@ export function PluggableStorage(options: PluggableStorageOptions): IStorageAsyn const connectPromise = wrapper.connect().then(() => { onReadyCb(); - // Start periodic flush of async storages - if (impressionCountsCache && (impressionCountsCache as ImpressionCountsCachePluggable).start) (impressionCountsCache as ImpressionCountsCachePluggable).start(); - if (uniqueKeysCache && (uniqueKeysCache as UniqueKeysCachePluggable).start) (uniqueKeysCache as UniqueKeysCachePluggable).start(); - - // If mode is not defined, it means that the synchronizer is running and so we don't have to record telemetry - if (telemetry && (telemetry as ITelemetryCacheAsync).recordConfig && mode) (telemetry as ITelemetryCacheAsync).recordConfig(); + // Start periodic flush of async storages if not running synchronizer (producer mode) + if (!isSyncronizer) { + if (impressionCountsCache && (impressionCountsCache as ImpressionCountsCachePluggable).start) (impressionCountsCache as ImpressionCountsCachePluggable).start(); + if (uniqueKeysCache && (uniqueKeysCache as UniqueKeysCachePluggable).start) (uniqueKeysCache as UniqueKeysCachePluggable).start(); + if (telemetry && (telemetry as ITelemetryCacheAsync).recordConfig) (telemetry as ITelemetryCacheAsync).recordConfig(); + } }).catch((e) => { e = e || new Error('Error connecting wrapper'); onReadyCb(e); @@ -102,9 +113,9 @@ export function PluggableStorage(options: PluggableStorageOptions): IStorageAsyn telemetry, uniqueKeys: uniqueKeysCache, - // Disconnect the underlying storage + // Stop periodic flush and disconnect the underlying storage destroy() { - return Promise.all([ + return Promise.all(isSyncronizer ? [] : [ impressionCountsCache && (impressionCountsCache as ImpressionCountsCachePluggable).stop && (impressionCountsCache as ImpressionCountsCachePluggable).stop(), uniqueKeysCache && (uniqueKeysCache as UniqueKeysCachePluggable).stop && (uniqueKeysCache as UniqueKeysCachePluggable).stop(), ]).then(() => wrapper.disconnect()); diff --git a/src/storages/types.ts b/src/storages/types.ts index 297310c8..7bf687ed 100644 --- a/src/storages/types.ts +++ b/src/storages/types.ts @@ -1,7 +1,6 @@ -import { MaybeThenable, IMetadata, ISplitFiltersValidation, ISplit } from '../dtos/types'; -import { ILogger } from '../logger/types'; -import { EventDataType, HttpErrors, HttpLatencies, ImpressionDataType, LastSync, Method, MethodExceptions, MethodLatencies, MultiMethodExceptions, MultiMethodLatencies, MultiConfigs, OperationType, StoredEventWithMetadata, StoredImpressionWithMetadata, StreamingEvent, UniqueKeysPayloadCs, UniqueKeysPayloadSs } from '../sync/submitters/types'; -import { SplitIO, ImpressionDTO, SDKMode } from '../types'; +import { MaybeThenable, ISplit } from '../dtos/types'; +import { EventDataType, HttpErrors, HttpLatencies, ImpressionDataType, LastSync, Method, MethodExceptions, MethodLatencies, MultiMethodExceptions, MultiMethodLatencies, MultiConfigs, OperationType, StoredEventWithMetadata, StoredImpressionWithMetadata, StreamingEvent, UniqueKeysPayloadCs, UniqueKeysPayloadSs, TelemetryUsageStatsPayload } from '../sync/submitters/types'; +import { SplitIO, ImpressionDTO, ISettings } from '../types'; /** * Interface of a pluggable storage wrapper. @@ -70,21 +69,21 @@ export interface IPluggableStorageWrapper { /** Integer operations */ /** - * Increments the number stored at `key` by `increment` (or 1 if `increment` is not provided), or set it to `increment` (or 1) if the value doesn't exist. + * Increments the number stored at `key` by `increment`, or set it to `increment` if the value doesn't exist. * * @function incr * @param {string} key Key to increment - * @param {number} increment Value to increment by + * @param {number} increment Value to increment by. Defaults to 1. * @returns {Promise} A promise that resolves with the value of key after the increment. The promise rejects if the operation fails, * for example, if there is a connection error or the key contains a string that can not be represented as integer. */ incr: (key: string, increment?: number) => Promise /** - * Decrements the number stored at `key` by `decrement` (or 1 if `decrement` is not provided), or set it to minus `decrement` (or minus 1) if the value doesn't exist. + * Decrements the number stored at `key` by `decrement`, or set it to minus `decrement` if the value doesn't exist. * * @function decr * @param {string} key Key to decrement - * @param {number} decrement Value to decrement by + * @param {number} decrement Value to decrement by. Defaults to 1. * @returns {Promise} A promise that resolves with the value of key after the decrement. The promise rejects if the operation fails, * for example, if there is a connection error or the key contains a string that can not be represented as integer. */ @@ -285,19 +284,29 @@ export interface ISegmentsCacheAsync extends ISegmentsCacheBase { /** Recorder storages (impressions, events and telemetry) */ export interface IImpressionsCacheBase { - // Consumer API method, used by impressions tracker, in standalone and consumer modes, to push impressions into the storage. + // Used by impressions tracker, in DEBUG and OPTIMIZED impression modes, to push impressions into the storage. track(data: ImpressionDTO[]): MaybeThenable } export interface IEventsCacheBase { - // Consumer API method, used by events tracker, in standalone and consumer modes, to push events into the storage. + // Used by events tracker to push events into the storage. track(data: SplitIO.EventData, size?: number): MaybeThenable } -/** Impressions and events cache for standalone mode (sync) */ +export interface IImpressionCountsCacheBase { + // Used by impressions tracker, in OPTIMIZED and NONE impression modes, to count impressions. + track(featureName: string, timeFrame: number, amount: number): void +} + +export interface IUniqueKeysCacheBase { + // Used by impressions tracker, in NONE impression mode, to track unique keys. + track(key: string, value: string): void +} -// Producer API methods for sync recorder storages, used by submitters in standalone mode to pop data and post it to Split BE. -export interface IRecorderCacheProducerSync { +/** Impressions and events cache for standalone and partial consumer modes (sync methods) */ + +// API methods for sync recorder storages, used by submitters in standalone mode to pop data and post it to Split BE. +export interface IRecorderCacheSync { // @TODO names are inconsistent with spec /* Checks if cache is empty. Returns true if the cache was just created or cleared */ isEmpty(): boolean @@ -307,23 +316,29 @@ export interface IRecorderCacheProducerSync { pop(toMerge?: T): T } - -export interface IImpressionsCacheSync extends IImpressionsCacheBase, IRecorderCacheProducerSync { +export interface IImpressionsCacheSync extends IImpressionsCacheBase, IRecorderCacheSync { track(data: ImpressionDTO[]): void /* Registers callback for full queue */ setOnFullQueueCb(cb: () => void): void } -export interface IEventsCacheSync extends IEventsCacheBase, IRecorderCacheProducerSync { +export interface IEventsCacheSync extends IEventsCacheBase, IRecorderCacheSync { track(data: SplitIO.EventData, size?: number): boolean /* Registers callback for full queue */ setOnFullQueueCb(cb: () => void): void } -/** Impressions and events cache for consumer and producer mode (async) */ +/* Named `ImpressionsCounter` in spec */ +export interface IImpressionCountsCacheSync extends IImpressionCountsCacheBase, IRecorderCacheSync> { } -// Producer API methods for async recorder storages, used by submitters in producer mode to pop data and post it to Split BE. -export interface IRecorderCacheProducerAsync { +export interface IUniqueKeysCacheSync extends IUniqueKeysCacheBase, IRecorderCacheSync { + setOnFullQueueCb(cb: () => void): void, +} + +/** Impressions and events cache for consumer and producer modes (async methods) */ + +// API methods for async recorder storages, used by submitters in producer mode (synchronizer) to pop data and post it to Split BE. +export interface IRecorderCacheAsync { /* returns the number of stored items */ count(): Promise /* removes the given number of items from the store. If not provided, it deletes all items */ @@ -332,43 +347,18 @@ export interface IRecorderCacheProducerAsync { popNWithMetadata(count: number): Promise } -export interface IImpressionsCacheAsync extends IImpressionsCacheBase, IRecorderCacheProducerAsync { +export interface IImpressionsCacheAsync extends IImpressionsCacheBase, IRecorderCacheAsync { // Consumer API method, used by impressions tracker (in standalone and consumer modes) to push data into. // The result promise can reject. track(data: ImpressionDTO[]): Promise } -export interface IEventsCacheAsync extends IEventsCacheBase, IRecorderCacheProducerAsync { +export interface IEventsCacheAsync extends IEventsCacheBase, IRecorderCacheAsync { // Consumer API method, used by events tracker (in standalone and consumer modes) to push data into. // The result promise cannot reject. track(data: SplitIO.EventData, size?: number): Promise } -/** - * Impression counts cache for impressions dedup in standalone and producer mode. - * Only in memory. Named `ImpressionsCounter` in spec. - */ -export interface IImpressionCountsCacheSync extends IRecorderCacheProducerSync> { -// Used by impressions tracker - track(featureName: string, timeFrame: number, amount: number): void - - // Used by impressions count submitter in standalone and producer mode - isEmpty(): boolean // check if cache is empty. Return true if the cache was just created or cleared. - pop(toMerge?: Record ): Record // pop cache data -} - -export interface IUniqueKeysCacheBase { - // Used by unique Keys tracker - track(key: string, value: string): void - - // Used by unique keys submitter in standalone and producer mode - isEmpty(): boolean // check if cache is empty. Return true if the cache was just created or cleared. - pop(): UniqueKeysPayloadSs | UniqueKeysPayloadCs // pop cache data - /* Registers callback for full queue */ - setOnFullQueueCb(cb: () => void): void, - clear(): void -} - /** * Telemetry storage interface for standalone and partial consumer modes. * Methods are sync because data is stored in memory. @@ -428,7 +418,7 @@ export interface ITelemetryEvaluationProducerSync { export interface ITelemetryStorageProducerSync extends ITelemetryInitProducerSync, ITelemetryRuntimeProducerSync, ITelemetryEvaluationProducerSync { } -export interface ITelemetryCacheSync extends ITelemetryStorageConsumerSync, ITelemetryStorageProducerSync { } +export interface ITelemetryCacheSync extends ITelemetryStorageConsumerSync, ITelemetryStorageProducerSync, IRecorderCacheSync { } /** * Telemetry storage interface for consumer mode. @@ -458,7 +448,7 @@ export interface IStorageBase< TSplitsCache extends ISplitsCacheBase, TSegmentsCache extends ISegmentsCacheBase, TImpressionsCache extends IImpressionsCacheBase, - TImpressionsCountCache extends IImpressionCountsCacheSync, + TImpressionsCountCache extends IImpressionCountsCacheBase, TEventsCache extends IEventsCacheBase, TTelemetryCache extends ITelemetryCacheSync | ITelemetryCacheAsync, TUniqueKeysCache extends IUniqueKeysCacheBase @@ -481,14 +471,14 @@ export interface IStorageSync extends IStorageBase< IImpressionCountsCacheSync, IEventsCacheSync, ITelemetryCacheSync, - IUniqueKeysCacheBase + IUniqueKeysCacheSync > { } export interface IStorageAsync extends IStorageBase< ISplitsCacheAsync, ISegmentsCacheAsync, IImpressionsCacheAsync | IImpressionsCacheSync, - IImpressionCountsCacheSync, + IImpressionCountsCacheBase, IEventsCacheAsync | IEventsCacheSync, ITelemetryCacheAsync | ITelemetryCacheSync, IUniqueKeysCacheBase @@ -499,21 +489,12 @@ export interface IStorageAsync extends IStorageBase< export type DataLoader = (storage: IStorageSync, matchingKey: string) => void export interface IStorageFactoryParams { - log: ILogger, - impressionsQueueSize?: number, - eventsQueueSize?: number, - optimize?: boolean /* whether create the `impressionCounts` cache (OPTIMIZED impression mode) or not (DEBUG impression mode) */, - mode: SDKMode, - impressionsMode?: string, - // ATM, only used by InLocalStorage - matchingKey?: string, /* undefined on server-side SDKs */ - splitFiltersValidation?: ISplitFiltersValidation, - - // This callback is invoked when the storage is ready to be used. Error-first callback style: if an error is passed, - // it means that the storge fail to connect and shouldn't be used. - // It is meant for emitting SDK_READY event in consumer mode, and for synchronizer to wait before using the storage. + settings: ISettings, + /** + * Error-first callback invoked when the storage is ready to be used. An error means that the storage failed to connect and shouldn't be used. + * It is meant for emitting SDK_READY event in consumer mode, and waiting before using the storage in the synchronizer. + */ onReadyCb: (error?: any) => void, - metadata: IMetadata, } export type StorageType = 'MEMORY' | 'LOCALSTORAGE' | 'REDIS' | 'PLUGGABLE'; diff --git a/src/storages/utils.ts b/src/storages/utils.ts new file mode 100644 index 00000000..2bf236e3 --- /dev/null +++ b/src/storages/utils.ts @@ -0,0 +1,78 @@ +// Shared utils for Redis and Pluggable storage + +import { IMetadata } from '../dtos/types'; +import { Method, StoredImpressionWithMetadata } from '../sync/submitters/types'; +import { ImpressionDTO, ISettings } from '../types'; +import { UNKNOWN } from '../utils/constants'; +import { MAX_LATENCY_BUCKET_COUNT } from './inMemory/TelemetryCacheInMemory'; +import { METHOD_NAMES } from './KeyBuilderSS'; + +export function metadataBuilder(settings: Pick): IMetadata { + return { + s: settings.version, + i: settings.runtime.ip || UNKNOWN, + n: settings.runtime.hostname || UNKNOWN, + }; +} + +// Converts impressions to be stored in Redis or pluggable storage. +export function impressionsToJSON(impressions: ImpressionDTO[], metadata: IMetadata): string[] { + return impressions.map(impression => { + const impressionWithMetadata: StoredImpressionWithMetadata = { + m: metadata, + i: { + k: impression.keyName, + b: impression.bucketingKey, + f: impression.feature, + t: impression.treatment, + r: impression.label, + c: impression.changeNumber, + m: impression.time, + pt: impression.pt, + } + }; + + return JSON.stringify(impressionWithMetadata); + }); +} + +// Utilities used by TelemetryCacheInRedis and TelemetryCachePluggable + +const REVERSE_METHOD_NAMES = Object.keys(METHOD_NAMES).reduce((acc, key) => { + acc[METHOD_NAMES[key as Method]] = key as Method; + return acc; +}, {} as Record); + + +export function parseMetadata(field: string): [metadata: string] | string { + const parts = field.split('/'); + if (parts.length !== 3) return `invalid subsection count. Expected 3, got: ${parts.length}`; + + const [s /* metadata.s */, n /* metadata.n */, i /* metadata.i */] = parts; + return [JSON.stringify({ s, n, i })]; +} + +export function parseExceptionField(field: string): [metadata: string, method: Method] | string { + const parts = field.split('/'); + if (parts.length !== 4) return `invalid subsection count. Expected 4, got: ${parts.length}`; + + const [s /* metadata.s */, n /* metadata.n */, i /* metadata.i */, m] = parts; + const method = REVERSE_METHOD_NAMES[m]; + if (!method) return `unknown method '${m}'`; + + return [JSON.stringify({ s, n, i }), method]; +} + +export function parseLatencyField(field: string): [metadata: string, method: Method, bucket: number] | string { + const parts = field.split('/'); + if (parts.length !== 5) return `invalid subsection count. Expected 5, got: ${parts.length}`; + + const [s /* metadata.s */, n /* metadata.n */, i /* metadata.i */, m, b] = parts; + const method = REVERSE_METHOD_NAMES[m]; + if (!method) return `unknown method '${m}'`; + + const bucket = parseInt(b); + if (isNaN(bucket) || bucket >= MAX_LATENCY_BUCKET_COUNT) return `invalid bucket. Expected a number between 0 and ${MAX_LATENCY_BUCKET_COUNT - 1}, got: ${b}`; + + return [JSON.stringify({ s, n, i }), method, bucket]; +} diff --git a/src/sync/submitters/__tests__/telemetrySubmitter.spec.ts b/src/sync/submitters/__tests__/telemetrySubmitter.spec.ts index bbe74a59..25ff16be 100644 --- a/src/sync/submitters/__tests__/telemetrySubmitter.spec.ts +++ b/src/sync/submitters/__tests__/telemetrySubmitter.spec.ts @@ -10,10 +10,15 @@ describe('Telemetry submitter', () => { const postMetricsUsage = jest.fn(() => Promise.resolve()); const postMetricsConfig = jest.fn(() => Promise.resolve()); const readinessGateCallbacks: Record void> = {}; + const settings = { + ...fullSettings, + core: { ...fullSettings.core, key: undefined }, // server-side -> storage.telemetry defined + scheduler: { ...fullSettings.scheduler, telemetryRefreshRate } + }; const params = { - settings: { ...fullSettings, scheduler: { ...fullSettings.scheduler, telemetryRefreshRate } }, + settings, splitApi: { postMetricsUsage, postMetricsConfig }, // @ts-ignore - storage: InMemoryStorageFactory({}), + storage: InMemoryStorageFactory({ settings }), platform: { now: () => 123 }, // by returning a fixed timestamp, all latencies are equal to 0 sdkReadinessManager: { incInternalReadyCbCount: jest.fn(), }, readiness: { @@ -27,21 +32,32 @@ describe('Telemetry submitter', () => { test('submits metrics/usage periodically', async () => { // @ts-ignore const telemetrySubmitter = telemetrySubmitterFactory(params) as ISyncTask; - const popLatenciesSpy = jest.spyOn(params.storage.telemetry!, 'popLatencies'); + const isEmptySpy = jest.spyOn(params.storage.telemetry!, 'isEmpty'); + const popSpy = jest.spyOn(params.storage.telemetry!, 'pop'); + + params.storage.telemetry?.addTag('tag1'); // add some data telemetrySubmitter.start(); expect(telemetrySubmitter.isRunning()).toEqual(true); // Submitter should be flagged as running expect(telemetrySubmitter.isExecuting()).toEqual(false); // but not executed immediatelly (first push window) - expect(popLatenciesSpy).toBeCalledTimes(0); + expect(popSpy).toBeCalledTimes(0); - // Await first push + // Await first periodic execution await new Promise(res => setTimeout(res, params.settings.scheduler.telemetryRefreshRate + 10)); - // after the first push, telemetry cache should have been used to create the request payload - expect(popLatenciesSpy).toBeCalledTimes(1); + // Telemetry cache is not empty, so data is popped and sent + expect(isEmptySpy).toBeCalledTimes(1); + expect(popSpy).toBeCalledTimes(1); expect(postMetricsUsage).toBeCalledWith(JSON.stringify({ - lS: {}, mL: {}, mE: {}, hE: {}, hL: {}, tR: 0, aR: 0, iQ: 0, iDe: 0, iDr: 0, spC: 0, seC: 0, skC: 0, eQ: 0, eD: 0, sE: [], t: [] + lS: {}, mL: {}, mE: {}, hE: {}, hL: {}, tR: 0, aR: 0, iQ: 0, iDe: 0, iDr: 0, spC: 0, seC: 0, skC: 0, eQ: 0, eD: 0, sE: [], t: ['tag1'] })); + // Await second periodic execution + await new Promise(res => setTimeout(res, params.settings.scheduler.telemetryRefreshRate + 10)); + // Telemetry cache is empty, so no data is popped and sent + expect(isEmptySpy).toBeCalledTimes(2); + expect(popSpy).toBeCalledTimes(1); + expect(postMetricsUsage).toBeCalledTimes(1); + expect(telemetrySubmitter.isRunning()).toEqual(true); telemetrySubmitter.stop(); expect(telemetrySubmitter.isRunning()).toEqual(false); @@ -61,7 +77,7 @@ describe('Telemetry submitter', () => { expect(recordTimeUntilReadySpy).toBeCalledTimes(1); expect(postMetricsConfig).toBeCalledWith(JSON.stringify({ - oM: 0, st: 'memory', aF: 0, rF: 0, sE: true, rR: { sp: 0.001, ms: 0.001, im: 0.001, ev: 0.001, te: 0.1 }, uO: { s: true, e: true, a: true, st: true, t: true }, iQ: 1, eQ: 1, iM: 0, iL: false, hP: false, tR: 0, tC: 0, nR: 0, t: [], i: ['NoopIntegration'], uC: 0 + oM: 0, st: 'memory', aF: 0, rF: 0, sE: true, rR: { sp: 0.001, se: 0.001, im: 0.001, ev: 0.001, te: 0.1 }, uO: { s: true, e: true, a: true, st: true, t: true }, iQ: 1, eQ: 1, iM: 0, iL: false, hP: false, tR: 0, tC: 0, nR: 0, t: [], i: ['NoopIntegration'], uC: 0 })); // Stop submitter, to not execute the 1st periodic metrics/usage POST diff --git a/src/sync/submitters/submitter.ts b/src/sync/submitters/submitter.ts index 2cf16d7c..da702e9e 100644 --- a/src/sync/submitters/submitter.ts +++ b/src/sync/submitters/submitter.ts @@ -1,6 +1,6 @@ import { syncTaskFactory } from '../syncTask'; import { ISyncTask } from '../types'; -import { IRecorderCacheProducerSync } from '../../storages/types'; +import { IRecorderCacheSync } from '../../storages/types'; import { ILogger } from '../../logger/types'; import { SUBMITTERS_PUSH, SUBMITTERS_PUSH_FAILS, SUBMITTERS_PUSH_RETRY } from '../../logger/constants'; import { IResponse } from '../../services/types'; @@ -11,7 +11,7 @@ import { IResponse } from '../../services/types'; export function submitterFactory( log: ILogger, postClient: (body: string) => Promise, - sourceCache: IRecorderCacheProducerSync, + sourceCache: IRecorderCacheSync, postRate: number, dataName: string, fromCacheToPayload?: (cacheData: T) => any, diff --git a/src/sync/submitters/submitterManager.ts b/src/sync/submitters/submitterManager.ts index 1e2a19d5..9313048c 100644 --- a/src/sync/submitters/submitterManager.ts +++ b/src/sync/submitters/submitterManager.ts @@ -16,7 +16,7 @@ export function submitterManagerFactory(params: ISdkFactoryContextSync): ISubmit const impressionCountsSubmitter = impressionCountsSubmitterFactory(params); if (impressionCountsSubmitter) submitters.push(impressionCountsSubmitter); const telemetrySubmitter = telemetrySubmitterFactory(params); - if (params.uniqueKeysTracker) submitters.push(uniqueKeysSubmitterFactory(params)); + if (params.storage.uniqueKeys) submitters.push(uniqueKeysSubmitterFactory(params)); return { // `onlyTelemetry` true if SDK is created with userConsent not GRANTED diff --git a/src/sync/submitters/telemetrySubmitter.ts b/src/sync/submitters/telemetrySubmitter.ts index a0b81ce7..041d1f26 100644 --- a/src/sync/submitters/telemetrySubmitter.ts +++ b/src/sync/submitters/telemetrySubmitter.ts @@ -1,7 +1,7 @@ -import { ISegmentsCacheSync, ISplitsCacheSync, ITelemetryCacheSync } from '../../storages/types'; +import { ITelemetryCacheSync } from '../../storages/types'; import { submitterFactory, firstPushWindowDecorator } from './submitter'; -import { TelemetryUsageStatsPayload, TelemetryConfigStatsPayload, TelemetryConfigStats } from './types'; -import { QUEUED, DEDUPED, DROPPED, CONSUMER_MODE, CONSUMER_ENUM, STANDALONE_MODE, CONSUMER_PARTIAL_MODE, STANDALONE_ENUM, CONSUMER_PARTIAL_ENUM, OPTIMIZED, DEBUG, NONE, DEBUG_ENUM, OPTIMIZED_ENUM, NONE_ENUM, CONSENT_GRANTED, CONSENT_DECLINED, CONSENT_UNKNOWN } from '../../utils/constants'; +import { TelemetryConfigStatsPayload, TelemetryConfigStats } from './types'; +import { CONSUMER_MODE, CONSUMER_ENUM, STANDALONE_MODE, CONSUMER_PARTIAL_MODE, STANDALONE_ENUM, CONSUMER_PARTIAL_ENUM, OPTIMIZED, DEBUG, NONE, DEBUG_ENUM, OPTIMIZED_ENUM, NONE_ENUM, CONSENT_GRANTED, CONSENT_DECLINED, CONSENT_UNKNOWN } from '../../utils/constants'; import { SDK_READY, SDK_READY_FROM_CACHE } from '../../readiness/constants'; import { ConsentStatus, ISettings, SDKMode } from '../../types'; import { base } from '../../utils/settingsValidation'; @@ -9,41 +9,6 @@ import { usedKeysMap } from '../../utils/inputValidation/apiKey'; import { timer } from '../../utils/timeTracker/timer'; import { ISdkFactoryContextSync } from '../../sdkFactory/types'; import { objectAssign } from '../../utils/lang/objectAssign'; -import { isStorageSync } from '../../trackers/impressionObserver/utils'; - -/** - * Converts data from telemetry cache into /metrics/usage request payload. - */ -export function telemetryCacheStatsAdapter(telemetry: ITelemetryCacheSync, splits?: ISplitsCacheSync, segments?: ISegmentsCacheSync) { - return { - isEmpty() { return false; }, // There is always data in telemetry cache - clear() { }, // No-op - - // @TODO consider moving inside telemetry cache for code size reduction - pop(): TelemetryUsageStatsPayload { - return { - lS: telemetry.getLastSynchronization(), - mL: telemetry.popLatencies(), - mE: telemetry.popExceptions(), - hE: telemetry.popHttpErrors(), - hL: telemetry.popHttpLatencies(), - tR: telemetry.popTokenRefreshes(), - aR: telemetry.popAuthRejections(), - iQ: telemetry.getImpressionStats(QUEUED), - iDe: telemetry.getImpressionStats(DEDUPED), - iDr: telemetry.getImpressionStats(DROPPED), - spC: splits && splits.getSplitNames().length, - seC: segments && segments.getRegisteredSegments().length, - skC: segments && segments.getKeysCount(), - sL: telemetry.getSessionLength(), - eQ: telemetry.getEventStats(QUEUED), - eD: telemetry.getEventStats(DROPPED), - sE: telemetry.popStreamingEvents(), - t: telemetry.popTags(), - }; - } - }; -} const OPERATION_MODE_MAP = { [STANDALONE_MODE]: STANDALONE_ENUM, @@ -131,7 +96,7 @@ export function telemetryCacheConfigAdapter(telemetry: ITelemetryCacheSync, sett * Submitter that periodically posts telemetry data */ export function telemetrySubmitterFactory(params: ISdkFactoryContextSync) { - const { storage: { splits, segments, telemetry }, platform: { now } } = params; + const { storage: { telemetry }, platform: { now } } = params; if (!telemetry || !now) return; // No submitter created if telemetry cache is not defined const { settings, settings: { log, scheduler: { telemetryRefreshRate } }, splitApi, readiness, sdkReadinessManager } = params; @@ -140,8 +105,7 @@ export function telemetrySubmitterFactory(params: ISdkFactoryContextSync) { const submitter = firstPushWindowDecorator( submitterFactory( log, splitApi.postMetricsUsage, - // @TODO cannot provide splits and segments cache if they are async, because `submitterFactory` expects a sync storage source - isStorageSync(params.settings) ? telemetryCacheStatsAdapter(telemetry, splits, segments) : telemetryCacheStatsAdapter(telemetry), + telemetry, telemetryRefreshRate, 'telemetry stats', undefined, 0, true ), telemetryRefreshRate diff --git a/src/sync/submitters/types.ts b/src/sync/submitters/types.ts index a516a68b..2fde3217 100644 --- a/src/sync/submitters/types.ts +++ b/src/sync/submitters/types.ts @@ -37,13 +37,15 @@ export type ImpressionCountsPayload = { }[] } +export type UniqueKeysItemSs = { + /** Split name */ + f: string + /** keyNames */ + ks: string[] +} + export type UniqueKeysPayloadSs = { - keys: { - /** Split name */ - f: string - /** keyNames */ - ks: string[] - }[] + keys: UniqueKeysItemSs[] } export type UniqueKeysPayloadCs = { @@ -74,6 +76,8 @@ export type StoredImpressionWithMetadata = { c: number, /** time */ m: number + /** previous time */ + pt?: number } } @@ -110,9 +114,9 @@ export type SEGMENT = 'se'; export type MY_SEGMENT = 'ms'; export type OperationType = SPLITS | IMPRESSIONS | IMPRESSIONS_COUNT | EVENTS | TELEMETRY | TOKEN | SEGMENT | MY_SEGMENT; -export type LastSync = Record -export type HttpErrors = Record -export type HttpLatencies = Record> +export type LastSync = Partial> +export type HttpErrors = Partial> +export type HttpLatencies = Partial>> export type TREATMENT = 't'; export type TREATMENTS = 'ts'; @@ -121,9 +125,9 @@ export type TREATMENTS_WITH_CONFIG = 'tcs'; export type TRACK = 'tr'; export type Method = TREATMENT | TREATMENTS | TREATMENT_WITH_CONFIG | TREATMENTS_WITH_CONFIG | TRACK; -export type MethodLatencies = Record>; +export type MethodLatencies = Partial>>; -export type MethodExceptions = Record; +export type MethodExceptions = Partial>; export type CONNECTION_ESTABLISHED = 0; export type OCCUPANCY_PRI = 10; diff --git a/src/trackers/__tests__/impressionsTracker.spec.ts b/src/trackers/__tests__/impressionsTracker.spec.ts index be3efd04..d5c40ee7 100644 --- a/src/trackers/__tests__/impressionsTracker.spec.ts +++ b/src/trackers/__tests__/impressionsTracker.spec.ts @@ -44,7 +44,7 @@ describe('Impressions Tracker', () => { fakeListener.logImpression.mockClear(); fakeIntegrationsManager.handleImpression.mockClear(); }); - + const strategy = strategyDebugFactory(impressionObserverCSFactory()); test('Tracker API', () => { diff --git a/src/trackers/__tests__/uniqueKeysTracker.spec.ts b/src/trackers/__tests__/uniqueKeysTracker.spec.ts index c7937268..a001af80 100644 --- a/src/trackers/__tests__/uniqueKeysTracker.spec.ts +++ b/src/trackers/__tests__/uniqueKeysTracker.spec.ts @@ -16,10 +16,10 @@ describe('Unique keys tracker', () => { clear: jest.fn(), }; - test('Should be able to track unique keys', () => { - + test('Should be able to track unique keys', () => { + const uniqueKeysTracker = uniqueKeysTrackerFactory(loggerMock, fakeUniqueKeysCache, fakeFilter); - + uniqueKeysTracker.track('key1', 'value1'); expect(fakeFilter.add).toBeCalledWith('key1','value1'); expect(fakeUniqueKeysCache.track).toBeCalledWith('key1','value1'); @@ -31,40 +31,40 @@ describe('Unique keys tracker', () => { uniqueKeysTracker.track('key2', 'value3'); expect(fakeFilter.add).toBeCalledWith('key2','value3'); expect(fakeUniqueKeysCache.track).toBeCalledTimes(3); - + fakeFilter.add = jest.fn(() => { return true; }); uniqueKeysTracker.track('key2', 'value4'); expect(fakeFilter.add).toBeCalledWith('key2','value4'); expect(fakeUniqueKeysCache.track).toBeCalledWith('key2','value4'); - + }); - - test('Unique keys filter cleaner', () => { - + + test('Unique keys filter cleaner', () => { + const refreshRate = 500; - + fakeFilter.refreshRate = refreshRate; - + const uniqueKeysTrackerWithRefresh = uniqueKeysTrackerFactory(loggerMock, fakeUniqueKeysCache, fakeFilter); - + setTimeout(() => { - + expect(fakeFilter.clear).toBeCalledTimes(1); - + setTimeout(() => { - + expect(fakeFilter.clear).toBeCalledTimes(2); uniqueKeysTrackerWithRefresh.stop(); - + setTimeout(() => { - + expect(fakeFilter.clear).toBeCalledTimes(2); }, refreshRate + 30); }, refreshRate + 30); - + }, refreshRate + 30); - + }); }); diff --git a/src/trackers/impressionObserver/utils.ts b/src/trackers/impressionObserver/utils.ts index 4ecccbe3..dc65be89 100644 --- a/src/trackers/impressionObserver/utils.ts +++ b/src/trackers/impressionObserver/utils.ts @@ -1,21 +1,6 @@ -import { CONSUMER_MODE, CONSUMER_PARTIAL_MODE, OPTIMIZED, PRODUCER_MODE, STANDALONE_MODE } from '../../utils/constants'; +import { CONSUMER_MODE, CONSUMER_PARTIAL_MODE } from '../../utils/constants'; import { ISettings } from '../../types'; -/** - * Checks if impressions previous time should be added or not. - */ -export function shouldAddPt(settings: ISettings) { - return [PRODUCER_MODE, STANDALONE_MODE, CONSUMER_PARTIAL_MODE].indexOf(settings.mode) > -1 ? true : false; -} - -/** - * Checks if it should dedupe impressions or not. - */ -export function shouldBeOptimized(settings: ISettings) { - if (!shouldAddPt(settings)) return false; - return settings.sync.impressionsMode === OPTIMIZED ? true : false; -} - /** * Storage is async if mode is consumer or partial consumer */ diff --git a/src/trackers/impressionsTracker.ts b/src/trackers/impressionsTracker.ts index c36edf3f..522b83db 100644 --- a/src/trackers/impressionsTracker.ts +++ b/src/trackers/impressionsTracker.ts @@ -31,9 +31,9 @@ export function impressionsTrackerFactory( const impressionsCount = impressions.length; const { impressionsToStore, impressionsToListener, deduped } = strategy.process(impressions); - + const impressionsToListenerCount = impressionsToListener.length; - + if ( impressionsToStore.length>0 ){ const res = impressionsCache.track(impressionsToStore); diff --git a/src/trackers/strategy/__tests__/strategyDebug.spec.ts b/src/trackers/strategy/__tests__/strategyDebug.spec.ts index 391db1a1..8ef5e71c 100644 --- a/src/trackers/strategy/__tests__/strategyDebug.spec.ts +++ b/src/trackers/strategy/__tests__/strategyDebug.spec.ts @@ -4,28 +4,28 @@ import { strategyDebugFactory } from '../strategyDebug'; import { impression1, impression2 } from './testUtils'; test('strategyDebug', () => { - + let augmentedImp1 = { ...impression1, pt: undefined }; let augmentedImp12 = { ...impression1, pt: impression1.time }; let augmentedImp2 = { ...impression2, pt: undefined }; - + let impressions = [impression1, impression2, {...impression1}]; let augmentedImpressions = [augmentedImp1, augmentedImp2, augmentedImp12]; - + const strategyDebugSS = strategyDebugFactory(impressionObserverSSFactory()); - + let { impressionsToStore, impressionsToListener, deduped } = strategyDebugSS.process(impressions); - + expect(impressionsToStore).toStrictEqual(augmentedImpressions); expect(impressionsToListener).toStrictEqual(augmentedImpressions); expect(deduped).toStrictEqual(0); - + const strategyDebugCS = strategyDebugFactory(impressionObserverCSFactory()); - + ({ impressionsToStore, impressionsToListener, deduped } = strategyDebugCS.process(impressions)); - + expect(impressionsToStore).toStrictEqual(augmentedImpressions); expect(impressionsToListener).toStrictEqual(augmentedImpressions); expect(deduped).toStrictEqual(0); - + }); diff --git a/src/trackers/strategy/__tests__/strategyOptimized.spec.ts b/src/trackers/strategy/__tests__/strategyOptimized.spec.ts index 880c3fc2..f13219b1 100644 --- a/src/trackers/strategy/__tests__/strategyOptimized.spec.ts +++ b/src/trackers/strategy/__tests__/strategyOptimized.spec.ts @@ -5,30 +5,30 @@ import { ImpressionCountsCacheInMemory } from '../../../storages/inMemory/Impres import { impression1, impression2 } from './testUtils'; test('strategyOptimized', () => { - + let augmentedImp1 = { ...impression1, pt: undefined }; let augmentedImp12 = { ...impression1, pt: impression1.time }; let augmentedImp13 = { ...impression1, pt: impression1.time }; let augmentedImp2 = { ...impression2, pt: undefined }; - + const impressionCountsCache = new ImpressionCountsCacheInMemory(); const impressions = [impression1, impression2, {...impression1}, {...impression1}]; const augmentedImpressions = [augmentedImp1, augmentedImp2, augmentedImp12, augmentedImp13]; - + const strategyOptimizedSS = strategyOptimizedFactory(impressionObserverSSFactory(), impressionCountsCache); - + let { impressionsToStore, impressionsToListener, deduped } = strategyOptimizedSS.process(impressions); expect(impressionsToStore).toStrictEqual([augmentedImp1, augmentedImp2]); expect(impressionsToListener).toStrictEqual(augmentedImpressions); expect(deduped).toStrictEqual(2); - + const strategyOptimizedCS = strategyOptimizedFactory(impressionObserverCSFactory(), impressionCountsCache); - + ({ impressionsToStore, impressionsToListener, deduped } = strategyOptimizedCS.process(impressions)); - + expect(impressionsToStore).toStrictEqual([augmentedImp1, augmentedImp2]); expect(impressionsToListener).toStrictEqual(augmentedImpressions); expect(deduped).toStrictEqual(2); - + }); diff --git a/src/trackers/strategy/strategyDebug.ts b/src/trackers/strategy/strategyDebug.ts index 386f5234..c6d29e8d 100644 --- a/src/trackers/strategy/strategyDebug.ts +++ b/src/trackers/strategy/strategyDebug.ts @@ -4,14 +4,14 @@ import { IStrategy } from '../types'; /** * Debug strategy for impressions tracker. Wraps impressions to store and adds previousTime if it corresponds - * + * * @param impressionsObserver impression observer. Previous time (pt property) is included in impression instances * @returns IStrategyResult */ export function strategyDebugFactory( impressionsObserver: IImpressionObserver ): IStrategy { - + return { process(impressions: ImpressionDTO[]) { impressions.forEach((impression) => { @@ -19,8 +19,8 @@ export function strategyDebugFactory( impression.pt = impressionsObserver.testAndSet(impression); }); return { - impressionsToStore: impressions, - impressionsToListener: impressions, + impressionsToStore: impressions, + impressionsToListener: impressions, deduped: 0 }; } diff --git a/src/trackers/strategy/strategyNone.ts b/src/trackers/strategy/strategyNone.ts index 4db99af8..0a2e75ef 100644 --- a/src/trackers/strategy/strategyNone.ts +++ b/src/trackers/strategy/strategyNone.ts @@ -1,32 +1,32 @@ -import { IImpressionCountsCacheSync } from '../../storages/types'; +import { IImpressionCountsCacheBase } from '../../storages/types'; import { ImpressionDTO } from '../../types'; import { IStrategy, IUniqueKeysTracker } from '../types'; /** * None strategy for impressions tracker. - * + * * @param impressionsCounter cache to save impressions count. impressions will be deduped (OPTIMIZED mode) - * @param uniqueKeysTracker unique keys tracker in charge of tracking the unique keys per split. + * @param uniqueKeysTracker unique keys tracker in charge of tracking the unique keys per split. * @returns IStrategyResult */ export function strategyNoneFactory( - impressionsCounter: IImpressionCountsCacheSync, + impressionsCounter: IImpressionCountsCacheBase, uniqueKeysTracker: IUniqueKeysTracker ): IStrategy { - + return { process(impressions: ImpressionDTO[]) { - impressions.forEach((impression) => { + impressions.forEach((impression) => { const now = Date.now(); // Increments impression counter per featureName impressionsCounter.track(impression.feature, now, 1); // Keep track by unique key uniqueKeysTracker.track(impression.keyName, impression.feature); }); - + return { - impressionsToStore: [], - impressionsToListener: impressions, + impressionsToStore: [], + impressionsToListener: impressions, deduped: 0 }; } diff --git a/src/trackers/strategy/strategyOptimized.ts b/src/trackers/strategy/strategyOptimized.ts index f188d964..ce1a9857 100644 --- a/src/trackers/strategy/strategyOptimized.ts +++ b/src/trackers/strategy/strategyOptimized.ts @@ -1,4 +1,4 @@ -import { IImpressionCountsCacheSync } from '../../storages/types'; +import { IImpressionCountsCacheBase } from '../../storages/types'; import { ImpressionDTO } from '../../types'; import { truncateTimeFrame } from '../../utils/time'; import { IImpressionObserver } from '../impressionObserver/types'; @@ -6,35 +6,35 @@ import { IStrategy } from '../types'; /** * Optimized strategy for impressions tracker. Wraps impressions to store and adds previousTime if it corresponds - * + * * @param impressionsObserver impression observer. previous time (pt property) is included in impression instances * @param impressionsCounter cache to save impressions count. impressions will be deduped (OPTIMIZED mode) * @returns IStrategyResult */ export function strategyOptimizedFactory( impressionsObserver: IImpressionObserver, - impressionsCounter: IImpressionCountsCacheSync, + impressionsCounter: IImpressionCountsCacheBase, ): IStrategy { - + return { process(impressions: ImpressionDTO[]) { const impressionsToStore: ImpressionDTO[] = []; impressions.forEach((impression) => { impression.pt = impressionsObserver.testAndSet(impression); - + const now = Date.now(); - + // Increments impression counter per featureName if (impression.pt) impressionsCounter.track(impression.feature, now, 1); - + // Checks if the impression should be added in queue to be sent if (!impression.pt || impression.pt < truncateTimeFrame(now)) { impressionsToStore.push(impression); } }); return { - impressionsToStore: impressionsToStore, - impressionsToListener: impressions, + impressionsToStore: impressionsToStore, + impressionsToListener: impressions, deduped: impressions.length - impressionsToStore.length }; } diff --git a/src/trackers/uniqueKeysTracker.ts b/src/trackers/uniqueKeysTracker.ts index 39352e7c..fe367c79 100644 --- a/src/trackers/uniqueKeysTracker.ts +++ b/src/trackers/uniqueKeysTracker.ts @@ -4,16 +4,16 @@ import { IUniqueKeysCacheBase } from '../storages/types'; import { IFilterAdapter, IUniqueKeysTracker } from './types'; const noopFilterAdapter = { - add() {return true;}, - contains() {return true;}, - clear() {} + add() { return true; }, + contains() { return true; }, + clear() { } }; /** * Trackes uniques keys * Unique Keys Tracker will be in charge of checking if the MTK was already sent to the BE in the last period - * or schedule to be sent; if not it will be added in an internal cache and sent in the next post. - * + * or schedule to be sent; if not it will be added in an internal cache and sent in the next post. + * * @param log Logger instance * @param uniqueKeysCache cache to save unique keys * @param filterAdapter filter adapter @@ -42,7 +42,7 @@ export function uniqueKeysTrackerFactory( stop(): void { clearInterval(intervalId); } - + }; } diff --git a/src/utils/inputValidation/__tests__/attribute.spec.ts b/src/utils/inputValidation/__tests__/attribute.spec.ts index a9d7f6e4..e0d3ca4b 100644 --- a/src/utils/inputValidation/__tests__/attribute.spec.ts +++ b/src/utils/inputValidation/__tests__/attribute.spec.ts @@ -3,7 +3,7 @@ import { loggerMock } from '../../../logger/__tests__/sdkLogger.mock'; test('INPUT VALIDATION for Attribute', () => { - + // @ts-ignore expect(validateAttribute(loggerMock, 2, 'dos', 'some_method_attrs')).toEqual(false); // It should be invalid if the attribute key is not a string expect(validateAttribute(loggerMock, '', 'empty', 'some_method_attrs')).toEqual(false); // It should be invalid if the attribute key is not a string diff --git a/src/utils/redis/RedisMock.ts b/src/utils/redis/RedisMock.ts index 97639342..93880329 100644 --- a/src/utils/redis/RedisMock.ts +++ b/src/utils/redis/RedisMock.ts @@ -12,9 +12,9 @@ const ASYNC_METHODS = ['rpush', 'hincrby']; const PIPELINE_METHODS = ['rpush', 'hincrby']; export class RedisMock { - + private pipelineMethods: any = { exec: jest.fn(asyncFunction) } - + constructor() { IDENTITY_METHODS.forEach(method => { this[method] = jest.fn(identityFunction); @@ -25,9 +25,9 @@ export class RedisMock { PIPELINE_METHODS.forEach(method => { this.pipelineMethods[method] = this[method]; }); - + this.pipeline = jest.fn(() => {return this.pipelineMethods;}); } - - + + } diff --git a/src/utils/settingsValidation/__tests__/index.spec.ts b/src/utils/settingsValidation/__tests__/index.spec.ts index 2b3f4918..8e0238c4 100644 --- a/src/utils/settingsValidation/__tests__/index.spec.ts +++ b/src/utils/settingsValidation/__tests__/index.spec.ts @@ -222,6 +222,17 @@ describe('settingsValidation', () => { expect(integrationsValidatorMock).toBeCalledWith(settings); }); + test('ignores key in server-side', () => { + const settings = settingsValidation({ + core: { + authorizationKey: 'dummy token', + key: 'ignored' + } + }, minimalSettingsParams); + + expect(settings.core.key).toBe(undefined); + }); + test('validates and sanitizes key and traffic type in client-side', () => { const clientSideValidationParams = { ...minimalSettingsParams, acceptKey: true, acceptTT: true }; diff --git a/src/utils/settingsValidation/index.ts b/src/utils/settingsValidation/index.ts index ee62d4f1..61e0f8d4 100644 --- a/src/utils/settingsValidation/index.ts +++ b/src/utils/settingsValidation/index.ts @@ -154,8 +154,8 @@ export function settingsValidation(config: unknown, validationParams: ISettingsV if (storage) withDefaults.storage = storage(withDefaults); // Validate key and TT (for client-side) + const maybeKey = withDefaults.core.key; if (validationParams.acceptKey) { - const maybeKey = withDefaults.core.key; // Although `key` is required in client-side, it can be omitted in LOCALHOST mode. In that case, the value `localhost_key` is used. if (withDefaults.mode === LOCALHOST_MODE && maybeKey === undefined) { withDefaults.core.key = 'localhost_key'; @@ -172,6 +172,10 @@ export function settingsValidation(config: unknown, validationParams: ISettingsV withDefaults.core.trafficType = validateTrafficType(log, maybeTT, 'Client instantiation'); } } + } else { + // On server-side, key is undefined and used to distinguish from client-side + if (maybeKey !== undefined) log.warn('Provided `key` is ignored in server-side SDK.'); // @ts-ignore + withDefaults.core.key = undefined; } // Current ip/hostname information