From d035440e93444784dcf5834814831bf938db086a Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Sun, 25 Feb 2024 13:59:28 -0800 Subject: [PATCH] cache l3 and add metric --- src/core/metrics.ts | 9 ++ src/dlob-subscriber/DLOBSubscriberIO.ts | 176 +++++++++++++++++++----- src/index.ts | 69 +++++++--- 3 files changed, 199 insertions(+), 55 deletions(-) diff --git a/src/core/metrics.ts b/src/core/metrics.ts index 09e920f..574c245 100644 --- a/src/core/metrics.ts +++ b/src/core/metrics.ts @@ -37,6 +37,7 @@ enum METRIC_TYPES { cache_miss_count = 'cache_miss_count', current_system_ts = 'current_system_ts', health_status = 'health_status', + incoming_requests_count = 'incoming_requests_count', } export enum HEALTH_STATUS { @@ -116,6 +117,13 @@ const cacheHitCounter = meter.createCounter(METRIC_TYPES.cache_hit_count, { description: 'Total redis cache hits', }); +const incomingRequestsCounter = meter.createCounter( + METRIC_TYPES.incoming_requests_count, + { + description: 'Total incoming requests', + } +); + const accountUpdatesCounter = meter.createCounter( METRIC_TYPES.account_updates_count, { @@ -225,6 +233,7 @@ export { endpointResponseTimeHistogram, gpaFetchDurationHistogram, responseStatusCounter, + incomingRequestsCounter, handleHealthCheck, setLastReceivedWsMsgTs, accountUpdatesCounter, diff --git a/src/dlob-subscriber/DLOBSubscriberIO.ts b/src/dlob-subscriber/DLOBSubscriberIO.ts index 2ef863f..64d2a54 100644 --- a/src/dlob-subscriber/DLOBSubscriberIO.ts +++ b/src/dlob-subscriber/DLOBSubscriberIO.ts @@ -15,7 +15,7 @@ import { l2WithBNToStrings, } from '../utils/utils'; -type wsMarketL2Args = { +type wsMarketArgs = { marketIndex: number; marketType: MarketType; marketName: string; @@ -36,7 +36,7 @@ export type wsMarketInfo = { }; export class DLOBSubscriberIO extends DLOBSubscriber { - public marketL2Args: wsMarketL2Args[] = []; + public marketArgs: wsMarketArgs[] = []; public lastSeenL2Formatted: Map>; redisClient: RedisClient; public killSwitchSlotDiffThreshold: number; @@ -73,7 +73,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber { ); const includeVamm = !isVariant(perpMarket.status, 'ammPaused'); - this.marketL2Args.push({ + this.marketArgs.push({ marketIndex: market.marketIndex, marketType: MarketType.PERP, marketName: market.marketName, @@ -85,7 +85,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber { }); } for (const market of config.spotMarketInfos) { - this.marketL2Args.push({ + this.marketArgs.push({ marketIndex: market.marketIndex, marketType: MarketType.SPOT, marketName: market.marketName, @@ -102,70 +102,75 @@ export class DLOBSubscriberIO extends DLOBSubscriber { override async updateDLOB(): Promise { await super.updateDLOB(); - for (const l2Args of this.marketL2Args) { + for (const marketArgs of this.marketArgs) { try { - this.getL2AndSendMsg(l2Args); + this.getL2AndSendMsg(marketArgs); + this.getL3AndSendMsg(marketArgs); } catch (error) { console.error(error); - console.log(`Error getting L2 ${l2Args.marketName}`); + console.log(`Error getting L2 ${marketArgs.marketName}`); } } } - getL2AndSendMsg(l2Args: wsMarketL2Args): void { - const grouping = l2Args.grouping; - const { marketName, ...l2FuncArgs } = l2Args; + getL2AndSendMsg(marketArgs: wsMarketArgs): void { + const grouping = marketArgs.grouping; + const { marketName, ...l2FuncArgs } = marketArgs; const l2 = this.getL2(l2FuncArgs); const slot = l2.slot; const lastMarketSlotAndTime = this.lastMarketSlotMap - .get(l2Args.marketType) - .get(l2Args.marketIndex); + .get(marketArgs.marketType) + .get(marketArgs.marketIndex); if (!lastMarketSlotAndTime) { this.lastMarketSlotMap - .get(l2Args.marketType) - .set(l2Args.marketIndex, { slot, ts: Date.now() }); + .get(marketArgs.marketType) + .set(marketArgs.marketIndex, { slot, ts: Date.now() }); } if (slot) { delete l2.slot; } - const marketType = isVariant(l2Args.marketType, 'perp') ? 'perp' : 'spot'; + const marketType = isVariant(marketArgs.marketType, 'perp') + ? 'perp' + : 'spot'; let l2Formatted: any; if (grouping) { const groupingBN = new BN(grouping); - l2Formatted = l2WithBNToStrings(groupL2(l2, groupingBN, l2Args.depth)); + l2Formatted = l2WithBNToStrings( + groupL2(l2, groupingBN, marketArgs.depth) + ); } else { l2Formatted = l2WithBNToStrings(l2); } - if (l2Args.updateOnChange) { + if (marketArgs.updateOnChange) { if ( this.lastSeenL2Formatted - .get(l2Args.marketType) - ?.get(l2Args.marketIndex) === JSON.stringify(l2Formatted) + .get(marketArgs.marketType) + ?.get(marketArgs.marketIndex) === JSON.stringify(l2Formatted) ) return; } this.lastSeenL2Formatted - .get(l2Args.marketType) - ?.set(l2Args.marketIndex, JSON.stringify(l2Formatted)); + .get(marketArgs.marketType) + ?.set(marketArgs.marketIndex, JSON.stringify(l2Formatted)); l2Formatted['marketName'] = marketName?.toUpperCase(); l2Formatted['marketType'] = marketType?.toLowerCase(); - l2Formatted['marketIndex'] = l2Args.marketIndex; + l2Formatted['marketIndex'] = marketArgs.marketIndex; l2Formatted['ts'] = Date.now(); l2Formatted['slot'] = slot; addOracletoResponse( l2Formatted, this.driftClient, - l2Args.marketType, - l2Args.marketIndex + marketArgs.marketType, + marketArgs.marketIndex ); addMarketSlotToResponse( l2Formatted, this.driftClient, - l2Args.marketType, - l2Args.marketIndex + marketArgs.marketType, + marketArgs.marketIndex ); // Check if slot diffs are too large for oracle @@ -200,12 +205,14 @@ export class DLOBSubscriberIO extends DLOBSubscriber { l2Formatted['marketSlot'] !== lastMarketSlotAndTime.slot ) { console.log( - `Updating market slot for ${l2Args.marketName} with slot ${l2Formatted['marketSlot']}` + `Updating market slot for ${marketArgs.marketName} with slot ${l2Formatted['marketSlot']}` ); - this.lastMarketSlotMap.get(l2Args.marketType).set(l2Args.marketIndex, { - slot: l2Formatted['marketSlot'], - ts: Date.now(), - }); + this.lastMarketSlotMap + .get(marketArgs.marketType) + .set(marketArgs.marketIndex, { + slot: l2Formatted['marketSlot'], + ts: Date.now(), + }); } const l2Formatted_depth100 = Object.assign({}, l2Formatted, { @@ -222,24 +229,121 @@ export class DLOBSubscriberIO extends DLOBSubscriber { }); this.redisClient.client.publish( - `orderbook_${marketType}_${l2Args.marketIndex}`, + `orderbook_${marketType}_${marketArgs.marketIndex}`, JSON.stringify(l2Formatted) ); this.redisClient.client.set( - `last_update_orderbook_${marketType}_${l2Args.marketIndex}`, + `last_update_orderbook_${marketType}_${marketArgs.marketIndex}`, JSON.stringify(l2Formatted_depth100) ); this.redisClient.client.set( - `last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_100`, + `last_update_orderbook_${marketType}_${marketArgs.marketIndex}_depth_100`, JSON.stringify(l2Formatted_depth100) ); this.redisClient.client.set( - `last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_20`, + `last_update_orderbook_${marketType}_${marketArgs.marketIndex}_depth_20`, JSON.stringify(l2Formatted_depth20) ); this.redisClient.client.set( - `last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_5`, + `last_update_orderbook_${marketType}_${marketArgs.marketIndex}_depth_5`, JSON.stringify(l2Formatted_depth5) ); } + + getL3AndSendMsg(marketArgs: wsMarketArgs): void { + const { marketName, ...l2FuncArgs } = marketArgs; + const l3 = this.getL3(l2FuncArgs); + const slot = l3.slot; + const lastMarketSlotAndTime = this.lastMarketSlotMap + .get(marketArgs.marketType) + .get(marketArgs.marketIndex); + if (!lastMarketSlotAndTime) { + this.lastMarketSlotMap + .get(marketArgs.marketType) + .set(marketArgs.marketIndex, { slot, ts: Date.now() }); + } + + if (slot) { + delete l3.slot; + } + const marketType = isVariant(marketArgs.marketType, 'perp') + ? 'perp' + : 'spot'; + + for (const key of Object.keys(l3)) { + for (const idx in l3[key]) { + const level = l3[key][idx]; + l3[key][idx] = { + ...level, + price: level.price.toString(), + size: level.size.toString(), + }; + } + } + + l3['marketName'] = marketName?.toUpperCase(); + l3['marketType'] = marketType?.toLowerCase(); + l3['marketIndex'] = marketArgs.marketIndex; + l3['ts'] = Date.now(); + l3['slot'] = slot; + addOracletoResponse( + l3, + this.driftClient, + marketArgs.marketType, + marketArgs.marketIndex + ); + addMarketSlotToResponse( + l3, + this.driftClient, + marketArgs.marketType, + marketArgs.marketIndex + ); + + // Check if slot diffs are too large for oracle + if ( + Math.abs(slot - parseInt(l3['oracleData']['slot'])) > + this.killSwitchSlotDiffThreshold + ) { + console.log(`Killing process due to slot diffs for market ${marketName}: + dlobProvider slot: ${slot} + oracle slot: ${l3['oracleData']['slot']} + `); + process.exit(1); + } + + // Check if times and slots are too different for market + const MAKRET_STALENESS_THRESHOLD = + marketType === 'perp' + ? PERP_MAKRET_STALENESS_THRESHOLD + : SPOT_MAKRET_STALENESS_THRESHOLD; + if ( + lastMarketSlotAndTime && + l3['marketSlot'] === lastMarketSlotAndTime.slot && + Date.now() - lastMarketSlotAndTime.ts > MAKRET_STALENESS_THRESHOLD + ) { + console.log(`Killing process due to same slot for market ${marketName} after > ${MAKRET_STALENESS_THRESHOLD}ms: + dlobProvider slot: ${slot} + market slot: ${l3['marketSlot']} + `); + process.exit(1); + } else if ( + lastMarketSlotAndTime && + l3['marketSlot'] !== lastMarketSlotAndTime.slot + ) { + console.log( + `Updating market slot for ${marketArgs.marketName} with slot ${l3['marketSlot']}` + ); + this.lastMarketSlotMap + .get(marketArgs.marketType) + .set(marketArgs.marketIndex, { + slot: l3['marketSlot'], + ts: Date.now(), + }); + } + + this.redisClient.client.set( + `last_update_orderbook_l3_${marketType}_${marketArgs.marketIndex}`, + JSON.stringify(l3) + ); + } } diff --git a/src/index.ts b/src/index.ts index 5092dd2..b5adc33 100644 --- a/src/index.ts +++ b/src/index.ts @@ -36,6 +36,7 @@ import { accountUpdatesCounter, cacheHitCounter, setLastReceivedWsMsgTs, + incomingRequestsCounter, runtimeSpecsGauge, } from './core/metrics'; import { handleResponseTime } from './core/middleware'; @@ -57,26 +58,29 @@ import { RedisClient } from './utils/redisClient'; require('dotenv').config(); // Reading in Redis env vars -const REDIS_HOSTS_ENV = (process.env.REDIS_HOSTS as string) || 'localhost'; -const REDIS_HOSTS = REDIS_HOSTS_ENV.includes(',') - ? REDIS_HOSTS_ENV.trim() - .replace(/^\[|\]$/g, '') - .split(/\s*,\s*/) - : [REDIS_HOSTS_ENV]; +const REDIS_HOSTS = process.env.REDIS_HOSTS?.replace(/^\[|\]$/g, '') + .split(',') + .map((host) => host.trim()) || ['localhost']; +const REDIS_PORTS = process.env.REDIS_PORTS?.replace(/^\[|\]$/g, '') + .split(',') + .map((port) => parseInt(port.trim(), 10)) || [6379]; +const REDIS_PASSWORDS_ENV = process.env.REDIS_PASSWORDS || "['']"; -const REDIS_PASSWORDS_ENV = (process.env.REDIS_PASSWORDS as string) || ''; -const REDIS_PASSWORDS = REDIS_PASSWORDS_ENV.includes(',') - ? REDIS_PASSWORDS_ENV.trim() - .replace(/^\[|\]$/g, '') - .split(/\s*,\s*/) - : [REDIS_PASSWORDS_ENV]; +let REDIS_PASSWORDS; + +if (REDIS_PASSWORDS_ENV.trim() === "['']") { + REDIS_PASSWORDS = [undefined]; +} else { + REDIS_PASSWORDS = REDIS_PASSWORDS_ENV.replace(/^\[|\]$/g, '') + .split(/\s*,\s*/) + .map((pwd) => pwd.replace(/(^'|'$)/g, '').trim()) + .map((pwd) => (pwd === '' ? undefined : pwd)); +} + +console.log('Redis Hosts:', REDIS_HOSTS); +console.log('Redis Ports:', REDIS_PORTS); +console.log('Redis Passwords:', REDIS_PASSWORDS); -const REDIS_PORTS_ENV = (process.env.REDIS_PORTS as string) || '6379'; -const REDIS_PORTS = REDIS_PORTS_ENV.includes(',') - ? REDIS_PORTS_ENV.trim() - .replace(/^\[|\]$/g, '') - .split(/\s*,\s*/) - : [REDIS_PORTS_ENV]; if ( REDIS_PORTS.length !== REDIS_PASSWORDS.length || REDIS_PORTS.length !== REDIS_HOSTS.length @@ -142,6 +146,7 @@ app.use((req, _res, next) => { req.url = '/'; } } + incomingRequestsCounter.add(1); next(); }); @@ -363,7 +368,7 @@ const main = async (): Promise => { redisClients.push( new RedisClient( REDIS_HOSTS[i], - REDIS_PORTS[i], + REDIS_PORTS[i].toString(), REDIS_PASSWORDS[i] || undefined ) ); @@ -1182,6 +1187,32 @@ const main = async (): Promise => { return; } + const marketTypeStr = getVariant(normedMarketType); + if (useRedis) { + const redisClient = ( + marketTypeStr === 'spot' ? spotMarketRedisMap : perpMarketRedisMap + ).get(normedMarketIndex).client; + const redisL3 = await redisClient.client.get( + `last_update_orderbook_l3_${marketTypeStr}_${normedMarketIndex}` + ); + if ( + redisL3 && + dlobProvider.getSlot() - parseInt(JSON.parse(redisL3).slot) < + SLOT_STALENESS_TOLERANCE + ) { + cacheHitCounter.add(1, { + miss: false, + }); + res.writeHead(200); + res.end(redisL3); + return; + } else { + cacheHitCounter.add(1, { + miss: true, + }); + } + } + const l3 = dlobSubscriber.getL3({ marketIndex: normedMarketIndex, marketType: normedMarketType,