trade/trade-drift-dlob
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor metrics, fix deps

Browse Source
This commit is contained in:
wphan
2025-07-15 11:56:56 -07:00
parent d19e33e840
commit 74eb54c164
13 changed files with 2278 additions and 2924 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
package.json
View File

@@ -17,7 +17,7 @@
"@project-serum/anchor": "^0.19.1-beta.1",
"@project-serum/serum": "^0.13.65",
"@pythnetwork/hermes-client": "^1.3.1",
"@solana/web3.js": "1.95.2",
"@solana/web3.js": "1.98.0",
"@triton-one/yellowstone-grpc": "^0.3.0",
"@types/redis": "^4.0.11",
"@types/ws": "^8.5.8",
@@ -31,13 +31,11 @@
"express": "^4.18.2",
"ioredis": "^5.4.1",
"morgan": "^1.10.0",
"prom-client": "^15.0.0",
"redis": "^4.6.10",
"response-time": "^2.3.2",
"rpc-websockets": "^10.0.0",
"rxjs": "^7.8.1",
"socket.io-redis": "^6.1.1",
"typescript": "4.5.4",
"typescript": "5.4.5",
"undici": "^6.16.1",
"winston": "^3.8.1",
"ws": "^8.14.2"
@@ -46,13 +44,13 @@
"@jest/globals": "^29.3.1",
"@types/jest": "^29.4.0",
"@types/k6": "^0.45.0",
"@typescript-eslint/eslint-plugin": "^4.28.0",
"@typescript-eslint/parser": "^4.28.0",
"@typescript-eslint/eslint-plugin": "6.21.0",
"@typescript-eslint/parser": "6.21.0",
"esbuild": "^0.24.0",
"eslint": "^7.29.0",
"eslint-config-prettier": "^8.3.0",
"eslint-plugin-prettier": "^3.4.0",
"husky": "^7.0.4",
"eslint": "8.57.0",
"eslint-config-prettier": "8.3.0",
"eslint-plugin-prettier": "3.4.0",
"jest": "^29.7.0",
"k6": "^0.0.0",
"prettier": "^2.4.1",
@@ -60,6 +58,10 @@
"ts-jest": "^29.1.0",
"ts-node": "^10.9.1"
},
"resolutions": {
"rpc-websockets": "10.0.0",
"@solana/web3.js": "1.98.0"
},
"scripts": {
"prepare": "husky install",
"build": "node esbuild.config.js",
@@ -102,8 +104,5 @@
"^@drift-labs/sdk$": "<rootDir>/drift-common/protocol/sdk/lib/node/index.js",
"^@drift/common$": "<rootDir>/drift-common/common-ts/lib/index.js"
}
},
"resolutions": {
"rpc-websockets": "^10.0.0"
}
}

95
src/core/healthCheck.ts Normal file
View File

@@ -0,0 +1,95 @@
export enum HEALTH_STATUS {
Ok = 0,
StaleBulkAccountLoader,
UnhealthySlotSubscriber,
LivenessTesting,
Restart,
}
/**
* Health check configuration
*/
const HEALTH_CHECK_CONFIG = {
CHECK_INTERVAL_MS: 2000,
// Maximum time allowed between slot updates
MAX_SLOT_STALENESS_MS: 5000,
// Minimum expected slot advancement rate (slots per second)
MIN_SLOT_RATE: 1,
} as const;
/**
* Tracks the health state of the slot subscriber
*/
type HealthState = {
lastSlot: number;
lastSlotTimestamp: number;
};
const globalHealthState: HealthState = {
lastSlot: -1,
lastSlotTimestamp: Date.now(),
};
/**
* Evaluates if the current state is healthy based on slot progression
*/
function evaluateHealth(currentSlot: number): {
isHealthy: boolean;
reason?: string;
} {
const now = Date.now();
// First health check
if (globalHealthState.lastSlot === -1) {
globalHealthState.lastSlot = currentSlot;
globalHealthState.lastSlotTimestamp = now;
return { isHealthy: true };
}
const timeDelta = now - globalHealthState.lastSlotTimestamp;
const slotDelta = currentSlot - globalHealthState.lastSlot;
// Update state if slot has progressed
if (currentSlot > globalHealthState.lastSlot) {
globalHealthState.lastSlot = currentSlot;
globalHealthState.lastSlotTimestamp = now;
}
// Check if slots are too stale
if (timeDelta > HEALTH_CHECK_CONFIG.MAX_SLOT_STALENESS_MS) {
return {
isHealthy: false,
reason: `No slot updates in ${timeDelta}ms (max ${HEALTH_CHECK_CONFIG.MAX_SLOT_STALENESS_MS}ms)`,
};
}
// Check if slot update rate is too low
const slotRate = (slotDelta / timeDelta) * 1000; // Convert to per second
if (slotRate < HEALTH_CHECK_CONFIG.MIN_SLOT_RATE) {
return {
isHealthy: false,
reason: `Slot update rate ${slotRate.toFixed(
2
)} slots/sec below minimum ${HEALTH_CHECK_CONFIG.MIN_SLOT_RATE}`,
};
}
return { isHealthy: true };
}
let healthStatus: HEALTH_STATUS = HEALTH_STATUS.Ok;
const setHealthStatus = (status: HEALTH_STATUS): void => {
healthStatus = status;
};
const getHealthStatus = (): HEALTH_STATUS => {
return healthStatus;
};
export {
evaluateHealth,
globalHealthState,
HEALTH_CHECK_CONFIG,
setHealthStatus,
getHealthStatus,
};

315
src/core/metrics.ts
View File

@@ -1,315 +0,0 @@
import { ObservableResult } from '@opentelemetry/api';
import { PrometheusExporter } from '@opentelemetry/exporter-prometheus';
import { logger } from '../utils/logger';
import {
ExplicitBucketHistogramAggregation,
InstrumentType,
MeterProvider,
View,
} from '@opentelemetry/sdk-metrics-base';
import { SlotSource } from '@drift-labs/sdk';
require('dotenv').config();
/**
* Creates {count} buckets of size {increment} starting from {start}. Each bucket stores the count of values within its "size".
* @param start
* @param increment
* @param count
* @returns
*/
const createHistogramBuckets = (
start: number,
increment: number,
count: number
) => {
return new ExplicitBucketHistogramAggregation(
Array.from(new Array(count), (_, i) => start + i * increment)
);
};
enum METRIC_TYPES {
runtime_specs = 'runtime_specs',
endpoint_response_times_histogram = 'endpoint_response_times_histogram',
endpoint_response_status = 'endpoint_response_status',
gpa_fetch_duration = 'gpa_fetch_duration',
last_ws_message_received_ts = 'last_ws_message_received_ts',
account_updates_count = 'account_updates_count',
cache_hit_count = 'cache_hit_count',
cache_miss_count = 'cache_miss_count',
current_system_ts = 'current_system_ts',
health_status = 'health_status',
incoming_requests_count = 'incoming_requests_count',
kinesis_records_sent = 'kinesis_records_sent',
}
export enum HEALTH_STATUS {
Ok = 0,
StaleBulkAccountLoader,
UnhealthySlotSubscriber,
LivenessTesting,
Restart,
}
const metricsPort =
parseInt(process.env.METRICS_PORT) || PrometheusExporter.DEFAULT_OPTIONS.port;
const { endpoint: defaultEndpoint } = PrometheusExporter.DEFAULT_OPTIONS;
const exporter = new PrometheusExporter(
{
port: metricsPort,
endpoint: defaultEndpoint,
},
() => {
logger.info(
`prometheus scrape endpoint started: http://localhost:${metricsPort}${defaultEndpoint}`
);
}
);
const meterName = 'dlob-meter';
const meterProvider = new MeterProvider({
views: [
new View({
instrumentName: METRIC_TYPES.endpoint_response_times_histogram,
instrumentType: InstrumentType.HISTOGRAM,
meterName,
aggregation: createHistogramBuckets(0, 20, 30),
}),
new View({
instrumentName: METRIC_TYPES.gpa_fetch_duration,
instrumentType: InstrumentType.HISTOGRAM,
meterName,
aggregation: createHistogramBuckets(0, 500, 20),
}),
],
});
meterProvider.addMetricReader(exporter);
const meter = meterProvider.getMeter(meterName);
const runtimeSpecsGauge = meter.createObservableGauge(
METRIC_TYPES.runtime_specs,
{
description: 'Runtime sepcification of this program',
}
);
let healthStatus: HEALTH_STATUS = HEALTH_STATUS.Ok;
const healthStatusGauge = meter.createObservableGauge(
METRIC_TYPES.health_status,
{
description: 'Health status of this program',
}
);
healthStatusGauge.addCallback((obs: ObservableResult) => {
obs.observe(healthStatus, {});
});
let lastWsMsgReceivedTs = 0;
const setLastReceivedWsMsgTs = (ts: number) => {
lastWsMsgReceivedTs = ts;
};
const lastWsReceivedTsGauge = meter.createObservableGauge(
METRIC_TYPES.last_ws_message_received_ts,
{
description: 'Timestamp of last received websocket message',
}
);
lastWsReceivedTsGauge.addCallback((obs: ObservableResult) => {
obs.observe(lastWsMsgReceivedTs, {});
});
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,
{
description: 'Total accounts update',
}
);
const currentSystemTsGauge = meter.createObservableGauge(
METRIC_TYPES.current_system_ts,
{
description: 'Timestamp of system at time of metric collection',
}
);
currentSystemTsGauge.addCallback((obs: ObservableResult) => {
obs.observe(Date.now(), {});
});
const endpointResponseTimeHistogram = meter.createHistogram(
METRIC_TYPES.endpoint_response_times_histogram,
{
description: 'Duration of endpoint responses',
unit: 'ms',
}
);
const gpaFetchDurationHistogram = meter.createHistogram(
METRIC_TYPES.gpa_fetch_duration,
{
description: 'Duration of GPA fetches',
unit: 'ms',
}
);
const responseStatusCounter = meter.createCounter(
METRIC_TYPES.endpoint_response_status,
{
description: 'Count of endpoint responses by status code',
}
);
const kinesisRecordsSentGauge = meter.createObservableGauge(
METRIC_TYPES.kinesis_records_sent,
{
description: 'Number of records successfully sent to Kinesis stream',
}
);
let kinesisRecordsSent: KinesisMetric = {
stream: '',
count: 0,
};
const setKinesisRecordsSent = (count: number, stream: string): void => {
kinesisRecordsSent = {
stream,
count,
};
};
kinesisRecordsSentGauge.addCallback((obs: ObservableResult) => {
obs.observe(kinesisRecordsSent.count, { stream: kinesisRecordsSent.stream });
});
/**
* Health check configuration
*/
const HEALTH_CHECK_CONFIG = {
CHECK_INTERVAL_MS: 2000,
// Maximum time allowed between slot updates
MAX_SLOT_STALENESS_MS: 5000,
// Minimum expected slot advancement rate (slots per second)
MIN_SLOT_RATE: 1,
} as const;
/**
* Tracks the health state of the slot subscriber
*/
type HealthState = {
lastSlot: number;
lastSlotTimestamp: number;
};
const globalHealthState: HealthState = {
lastSlot: -1,
lastSlotTimestamp: Date.now(),
};
/**
* Evaluates if the current state is healthy based on slot progression
*/
function evaluateHealth(currentSlot: number): {
isHealthy: boolean;
reason?: string;
} {
const now = Date.now();
// First health check
if (globalHealthState.lastSlot === -1) {
globalHealthState.lastSlot = currentSlot;
globalHealthState.lastSlotTimestamp = now;
return { isHealthy: true };
}
const timeDelta = now - globalHealthState.lastSlotTimestamp;
const slotDelta = currentSlot - globalHealthState.lastSlot;
// Update state if slot has progressed
if (currentSlot > globalHealthState.lastSlot) {
globalHealthState.lastSlot = currentSlot;
globalHealthState.lastSlotTimestamp = now;
}
// Check if slots are too stale
if (timeDelta > HEALTH_CHECK_CONFIG.MAX_SLOT_STALENESS_MS) {
return {
isHealthy: false,
reason: `No slot updates in ${timeDelta}ms (max ${HEALTH_CHECK_CONFIG.MAX_SLOT_STALENESS_MS}ms)`,
};
}
// Check if slot update rate is too low
const slotRate = (slotDelta / timeDelta) * 1000; // Convert to per second
if (slotRate < HEALTH_CHECK_CONFIG.MIN_SLOT_RATE) {
return {
isHealthy: false,
reason: `Slot update rate ${slotRate.toFixed(
2
)} slots/sec below minimum ${HEALTH_CHECK_CONFIG.MIN_SLOT_RATE}`,
};
}
return { isHealthy: true };
}
/**
* Middleware that checks the health of the slot subscriber.
* Health is determined by:
* 1. Regular slot updates
* 2. Minimum slot update rate
*/
const handleHealthCheck = (slotSource: SlotSource) => {
return async (_req, res, _next) => {
const currentSlot = slotSource.getSlot();
const { isHealthy, reason } = evaluateHealth(currentSlot);
if (!isHealthy) {
logger.error(
`Unhealthy: ${reason}, ` +
`Details: currentSlot=${currentSlot}, ` +
`lastSlot=${globalHealthState.lastSlot}, ` +
`timeSinceLastSlot=${
Date.now() - globalHealthState.lastSlotTimestamp
}ms`
);
setHealthStatus(HEALTH_STATUS.UnhealthySlotSubscriber);
res.writeHead(500);
res.end('NOK');
return;
}
setHealthStatus(HEALTH_STATUS.Ok);
res.writeHead(200);
res.end('OK');
};
};
export const setHealthStatus = (status: HEALTH_STATUS): void => {
healthStatus = status;
};
interface KinesisMetric {
stream: string;
count: number;
}
export {
endpointResponseTimeHistogram,
gpaFetchDurationHistogram,
responseStatusCounter,
incomingRequestsCounter,
handleHealthCheck,
setLastReceivedWsMsgTs,
accountUpdatesCounter,
cacheHitCounter,
runtimeSpecsGauge,
setKinesisRecordsSent,
};

209
src/core/metricsV2.ts Normal file
View File

@@ -0,0 +1,209 @@
import { Meter } from '@opentelemetry/api-metrics';
import {
ExplicitBucketHistogramAggregation,
MeterProvider,
View,
} from '@opentelemetry/sdk-metrics-base';
import { PrometheusExporter } from '@opentelemetry/exporter-prometheus';
import { logger } from '../utils/logger';
import { PublicKey } from '@solana/web3.js';
import { UserAccount } from '@drift-labs/sdk';
import {
BatchObservableResult,
Attributes,
ObservableGauge,
Histogram,
Counter,
} from '@opentelemetry/api';
/**
* RuntimeSpec is the attributes of the runtime environment, used to
* distinguish this metric set from others
*/
export type RuntimeSpec = {
rpcEndpoint: string;
driftEnv: string;
commit: string;
driftPid: string;
walletAuthority: string;
};
export function metricAttrFromUserAccount(
userAccountKey: PublicKey,
ua: UserAccount
): any {
return {
subaccount_id: ua.subAccountId,
public_key: userAccountKey.toBase58(),
authority: ua.authority.toBase58(),
delegate: ua.delegate.toBase58(),
};
}
/**
* Creates {count} buckets of size {increment} starting from {start}. Each bucket stores the count of values within its "size".
* @param start
* @param increment
* @param count
* @returns
*/
export function createHistogramBuckets(
start: number,
increment: number,
count: number
) {
return new ExplicitBucketHistogramAggregation(
Array.from(new Array(count), (_, i) => start + i * increment)
);
}
export class GaugeValue {
private latestGaugeValues: Map<string, number>;
private gauge: ObservableGauge;
constructor(gauge: ObservableGauge) {
this.gauge = gauge;
this.latestGaugeValues = new Map<string, number>();
}
setLatestValue(value: number, attributes: Attributes) {
const attributesStr = JSON.stringify(attributes);
this.latestGaugeValues.set(attributesStr, value);
}
getLatestValue(attributes: Attributes): number | undefined {
const attributesStr = JSON.stringify(attributes);
return this.latestGaugeValues.get(attributesStr);
}
getGauge(): ObservableGauge {
return this.gauge;
}
entries(): IterableIterator<[string, number]> {
return this.latestGaugeValues.entries();
}
}
export class HistogramValue {
private histogram: Histogram;
constructor(histogram: Histogram) {
this.histogram = histogram;
}
record(value: number, attributes: Attributes) {
this.histogram.record(value, attributes);
}
}
export class CounterValue {
private counter: Counter;
constructor(counter: Counter) {
this.counter = counter;
}
add(value: number, attributes: Attributes) {
this.counter.add(value, attributes);
}
}
export class Metrics {
private exporter: PrometheusExporter;
private meterProvider: MeterProvider;
private meters: Map<string, Meter>;
private gauges: Array<GaugeValue>;
private defaultMeterName: string;
constructor(meterName: string, views?: Array<View>, metricsPort?: number) {
const { endpoint: defaultEndpoint, port: defaultPort } =
PrometheusExporter.DEFAULT_OPTIONS;
const port = metricsPort || defaultPort;
this.exporter = new PrometheusExporter(
{
port: port,
endpoint: defaultEndpoint,
},
() => {
logger.info(
`prometheus scrape endpoint started: http://localhost:${port}${defaultEndpoint}`
);
}
);
this.meterProvider = new MeterProvider({ views });
this.meterProvider.addMetricReader(this.exporter);
this.gauges = new Array<GaugeValue>();
this.meters = new Map<string, Meter>();
this.defaultMeterName = meterName;
this.getMeter(this.defaultMeterName);
}
getMeter(name: string): Meter {
if (this.meters.has(name)) {
return this.meters.get(name) as Meter;
} else {
const meter = this.meterProvider.getMeter(name);
this.meters.set(name, meter);
return meter;
}
}
addGauge(
metricName: string,
description: string,
meterName?: string
): GaugeValue {
const meter = this.getMeter(meterName ?? this.defaultMeterName);
const newGauge = meter.createObservableGauge(metricName, {
description: description,
});
const gauge = new GaugeValue(newGauge);
this.gauges.push(gauge);
return gauge;
}
addHistogram(
metricName: string,
description: string,
meterName?: string
): HistogramValue {
const meter = this.getMeter(meterName ?? this.defaultMeterName);
return new HistogramValue(
meter.createHistogram(metricName, {
description: description,
})
);
}
addCounter(
metricName: string,
description: string,
meterName?: string
): CounterValue {
const meter = this.getMeter(meterName ?? this.defaultMeterName);
return new CounterValue(
meter.createCounter(metricName, {
description: description,
})
);
}
/**
* Finalizes the observables by adding the batch observable callback to each meter.
* Must call this before using this Metrics object
*/
finalizeObservables() {
for (const meter of this.meters.values()) {
meter.addBatchObservableCallback(
(observerResult: BatchObservableResult) => {
for (const gauge of this.gauges) {
for (const [attributesStr, value] of gauge.entries()) {
const attributes = JSON.parse(attributesStr);
observerResult.observe(gauge.getGauge(), value, attributes);
}
}
},
this.gauges.map((gauge) => gauge.getGauge())
);
}
}
}

77
src/core/middleware.ts
View File

@@ -1,15 +1,20 @@
import { Request, Response } from 'express';
import responseTime = require('response-time');
import {
endpointResponseTimeHistogram,
responseStatusCounter,
} from './metrics';
import { MEASURED_ENDPOINTS } from '../utils/constants';
import { SlotSource } from '@drift-labs/sdk';
import {
evaluateHealth,
globalHealthState,
setHealthStatus,
HEALTH_STATUS,
} from './healthCheck';
import { logger } from '../utils/logger';
import { GaugeValue } from './metricsV2';
export const handleResponseTime = (
measuredEndpoints: string[] = MEASURED_ENDPOINTS
) => {
return responseTime((req: Request, res: Response, time: number) => {
return responseTime((req: Request, _res: Response, _time: number) => {
const endpoint = req.path;
if (!measuredEndpoints.includes(endpoint)) {
@@ -19,14 +24,60 @@ export const handleResponseTime = (
// return;
//}
responseStatusCounter.add(1, {
endpoint,
status: res.statusCode,
});
// TODO: add these back if want to profile response times
const responseTimeMs = time;
endpointResponseTimeHistogram.record(responseTimeMs, {
endpoint,
});
// responseStatusCounter.add(1, {
// endpoint,
// status: res.statusCode,
// });
// const responseTimeMs = time;
// endpointResponseTimeHistogram.record(responseTimeMs, {
// endpoint,
// });
});
};
/**
* Middleware that checks the health of the slot subscriber.
* Health is determined by:
* 1. Regular slot updates
* 2. Minimum slot update rate
*/
export const handleHealthCheck = (
slotSource: SlotSource,
healthCheckGauge?: GaugeValue
) => {
return async (_req, res, _next) => {
const currentSlot = slotSource.getSlot();
const { isHealthy, reason } = evaluateHealth(currentSlot);
if (!isHealthy) {
logger.error(
`Unhealthy: ${reason}, ` +
`Details: currentSlot=${currentSlot}, ` +
`lastSlot=${globalHealthState.lastSlot}, ` +
`timeSinceLastSlot=${
Date.now() - globalHealthState.lastSlotTimestamp
}ms`
);
if (healthCheckGauge) {
healthCheckGauge.setLatestValue(
Number(HEALTH_STATUS.UnhealthySlotSubscriber),
{}
);
}
setHealthStatus(HEALTH_STATUS.UnhealthySlotSubscriber);
res.writeHead(500);
res.end('NOK');
return;
}
if (healthCheckGauge) {
healthCheckGauge.setLatestValue(Number(HEALTH_STATUS.Ok), {});
}
setHealthStatus(HEALTH_STATUS.Ok);
res.writeHead(200);
res.end('OK');
};
};

6
src/dlob-subscriber/DLOBSubscriberIO.ts
View File

@@ -28,8 +28,9 @@ import {
parsePositiveIntArray,
publishGroupings,
} from '../utils/utils';
import { setHealthStatus, HEALTH_STATUS } from '../core/metrics';
import { OffloadQueue } from '../utils/offload';
import { setHealthStatus, HEALTH_STATUS } from '../core/healthCheck';
import { CounterValue } from '../core/metricsV2';
export type wsMarketArgs = {
marketIndex: number;
@@ -84,6 +85,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
redisClient: RedisClient;
indicativeQuotesRedisClient?: RedisClient;
enableOffloadQueue?: boolean;
offloadQueueCounter?: CounterValue;
perpMarketInfos: wsMarketInfo[];
spotMarketInfos: wsMarketInfo[];
spotMarketSubscribers: SubscriberLookup;
@@ -99,7 +101,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
this.enableOffload = config.enableOffloadQueue || false;
if (this.enableOffload) {
this.offloadQueue = OffloadQueue();
this.offloadQueue = OffloadQueue(config.offloadQueueCounter);
}
// Set up appropriate maps

64
src/index.ts
View File

@@ -28,15 +28,8 @@ import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
import { logger, setLogLevel } from './utils/logger';
import * as http from 'http';
import {
handleHealthCheck,
accountUpdatesCounter,
cacheHitCounter,
setLastReceivedWsMsgTs,
incomingRequestsCounter,
runtimeSpecsGauge,
} from './core/metrics';
import { handleResponseTime } from './core/middleware';
import { Metrics } from './core/metricsV2';
import { handleHealthCheck } from './core/middleware';
import {
errorHandler,
normalizeBatchQueryParams,
@@ -93,12 +86,40 @@ const hermesUrl = process.env.HERMES_ENDPOINT;
const pythLazerDriftToken = process.env.PYTH_LAZER_DRIFT_TOKEN;
const pythLazerEndpoint = process.env.PYTH_LAZER_ENDPOINT;
const metricsPort = process.env.METRICS_PORT
? parseInt(process.env.METRICS_PORT)
: 9464;
const logFormat =
':remote-addr - :remote-user [:date[clf]] ":method :url HTTP/:http-version" :status :res[content-length] ":referrer" ":user-agent" :req[x-forwarded-for]';
const logHttp = morgan(logFormat, {
skip: (_req, res) => res.statusCode <= 500,
});
// init metrics
const metricsV2 = new Metrics('dlob-publisher', undefined, metricsPort);
const healthStatusGauge = metricsV2.addGauge(
'health_status',
'Health check status'
);
const accountUpdatesCounter = metricsV2.addCounter(
'account_updates_count',
'Total accounts update'
);
const cacheHitCounter = metricsV2.addCounter(
'cache_hit_count',
'Total cache hit'
);
const lastWsReceivedTsGauge = metricsV2.addGauge(
'last_ws_message_received_ts',
'Timestamp of last received websocket message'
);
const incomingRequestsCounter = metricsV2.addCounter(
'incoming_requests_count',
'Total incoming requests'
);
metricsV2.finalizeObservables();
let driftClient: DriftClient;
const app = express();
@@ -106,7 +127,6 @@ app.use(cors({ origin: '*' }));
app.use(compression());
app.set('trust proxy', 1);
app.use(logHttp);
app.use(handleResponseTime());
// strip off /dlob, if the request comes from exchange history server LB
app.use((req, _res, next) => {
@@ -116,21 +136,10 @@ app.use((req, _res, next) => {
req.url = '/';
}
}
incomingRequestsCounter.add(1);
incomingRequestsCounter.add(1, {});
next();
});
// Metrics defined here
const bootTimeMs = Date.now();
runtimeSpecsGauge.addCallback((obs) => {
obs.observe(bootTimeMs, {
commit: commitHash,
driftEnv,
rpcEndpoint: endpoint,
wsEndpoint: wsEndpoint,
});
});
app.use(errorHandler);
const server = http.createServer(app);
@@ -179,7 +188,6 @@ const main = async (): Promise<void> => {
delistedMarketSetting: DelistedMarketSetting.Discard,
});
let updatesReceivedTotal = 0;
const orderSubscriber = new OrderSubscriber({
driftClient,
subscriptionConfig: {
@@ -192,10 +200,8 @@ const main = async (): Promise<void> => {
orderSubscriber.eventEmitter.on(
'updateReceived',
(_pubkey: PublicKey, _slot: number, _dataType: 'raw' | 'decoded') => {
setLastReceivedWsMsgTs(Date.now());
// eslint-disable-next-line @typescript-eslint/no-unused-vars
updatesReceivedTotal++;
accountUpdatesCounter.add(1);
lastWsReceivedTsGauge.setLatestValue(Date.now(), {});
accountUpdatesCounter.add(1, {});
}
);
@@ -265,9 +271,9 @@ const main = async (): Promise<void> => {
}
};
app.get('/health', handleHealthCheck(dlobProvider));
app.get('/health', handleHealthCheck(dlobProvider, healthStatusGauge));
app.get('/startup', handleStartup);
app.get('/', handleHealthCheck(dlobProvider));
app.get('/', handleHealthCheck(dlobProvider, healthStatusGauge));
app.get('/priorityFees', async (req, res, next) => {
try {

111
src/publishers/dlobPublisher.ts
View File

@@ -43,9 +43,9 @@ import {
} from '../dlobProvider';
import FEATURE_FLAGS from '../utils/featureFlags';
import express, { Response, Request } from 'express';
import { handleHealthCheck } from '../core/metrics';
import { handleHealthCheck } from '../core/middleware';
import { setGlobalDispatcher, Agent } from 'undici';
import { register, Gauge } from 'prom-client';
import { Metrics } from '../core/metricsV2';
setGlobalDispatcher(
new Agent({
@@ -57,35 +57,34 @@ require('dotenv').config();
const stateCommitment: Commitment = 'confirmed';
const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
const commitHash = process.env.COMMIT;
const metricsPort = process.env.METRICS_PORT
? parseInt(process.env.METRICS_PORT)
: 9464;
const REDIS_CLIENT = process.env.REDIS_CLIENT || 'DLOB';
// Set up express for health checks
const app = express();
// metrics
const dlobSlotGauge = new Gauge({
name: 'dlob_slot',
help: 'Last updated slot of DLOB',
labelNames: [
'marketIndex',
'marketType',
'marketName',
'redisPrefix',
'redisClient',
],
});
const oracleSlotGauge = new Gauge({
name: 'oracle_slot',
help: 'Last updated slot of oracle',
labelNames: [
'marketIndex',
'marketType',
'marketName',
'redisPrefix',
'redisClient',
],
});
// init metrics
const metricsV2 = new Metrics('dlob-publisher', undefined, metricsPort);
const healthStatusGauge = metricsV2.addGauge(
'health_status',
'Health check status'
);
const dlobSlotGauge = metricsV2.addGauge(
'dlob_slot',
'Last updated slot of DLOB'
);
const oracleSlotGauge = metricsV2.addGauge(
'oracle_slot',
'Last updated slot of oracle'
);
const kinesisRecordsSentCounter = metricsV2.addCounter(
'kinesis_records_sent',
'Number of records sent to Kinesis'
);
metricsV2.finalizeObservables();
//@ts-ignore
const sdkConfig = initialize({ env: process.env.ENV });
@@ -528,6 +527,7 @@ const main = async () => {
protectedMakerView: false,
indicativeQuotesRedisClient: indicativeRedisClient,
enableOffloadQueue,
offloadQueueCounter: kinesisRecordsSentCounter,
});
await dlobSubscriberIndicative.subscribe();
@@ -555,41 +555,32 @@ const main = async () => {
const oracleDataAndSlot = driftClient.getOracleDataForPerpMarket(
market.marketIndex
);
dlobSlotGauge.set(
{
marketIndex: market.marketIndex,
marketType: 'perp',
marketName: market.marketName,
redisClient: REDIS_CLIENT,
redisPrefix: RedisClientPrefix[REDIS_CLIENT],
},
slot
);
oracleSlotGauge.set(
{
marketIndex: market.marketIndex,
marketType: 'perp',
marketName: market.marketName,
redisClient: REDIS_CLIENT,
redisPrefix: RedisClientPrefix[REDIS_CLIENT],
},
oracleDataAndSlot.slot.toNumber()
);
dlobSlotGauge.setLatestValue(slot, {
marketIndex: market.marketIndex,
marketType: 'perp',
marketName: market.marketName,
redisClient: REDIS_CLIENT,
redisPrefix: RedisClientPrefix[REDIS_CLIENT],
});
oracleSlotGauge.setLatestValue(oracleDataAndSlot.slot.toNumber(), {
marketIndex: market.marketIndex,
marketType: 'perp',
marketName: market.marketName,
redisClient: REDIS_CLIENT,
redisPrefix: RedisClientPrefix[REDIS_CLIENT],
});
});
spotMarketInfos.forEach((market) => {
const oracleDataAndSlot = driftClient.getOracleDataForSpotMarket(
market.marketIndex
);
dlobSlotGauge.set(
{
marketIndex: market.marketIndex,
marketType: 'spot',
marketName: market.marketName,
redisClient: REDIS_CLIENT,
redisPrefix: RedisClientPrefix[REDIS_CLIENT],
},
oracleDataAndSlot.slot.toNumber()
);
dlobSlotGauge.setLatestValue(oracleDataAndSlot.slot.toNumber(), {
marketIndex: market.marketIndex,
marketType: 'spot',
marketName: market.marketName,
redisClient: REDIS_CLIENT,
redisPrefix: RedisClientPrefix[REDIS_CLIENT],
});
});
}, 10_000);
@@ -646,15 +637,9 @@ const main = async () => {
}
};
app.get('/debug', handleDebug);
app.get('/metrics', async (req, res) => {
res.set('Content-Type', register.contentType);
res.end(await register.metrics());
});
app.get('/health', handleHealthCheck(dlobProvider));
app.get('/health', handleHealthCheck(slotSource, healthStatusGauge));
app.get('/startup', handleStartup);
app.get('/', handleHealthCheck(dlobProvider));
app.get('/', handleHealthCheck(slotSource, healthStatusGauge));
const server = app.listen(8080);
// Default keepalive is 5s, since the AWS ALB timeout is 60 seconds, clients

1
src/publishers/priorityFeesPublisher.ts
View File

@@ -15,7 +15,6 @@ import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
import { logger, setLogLevel } from '../utils/logger';
import { sleep } from '../utils/utils';
import express from 'express';
// import { handleHealthCheck } from '../core/metrics';
import { setGlobalDispatcher, Agent } from 'undici';
setGlobalDispatcher(

56
src/serverLite.ts
View File

@@ -18,15 +18,10 @@ import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
import { logger, setLogLevel } from './utils/logger';
import * as http from 'http';
import {
handleHealthCheck,
cacheHitCounter,
incomingRequestsCounter,
runtimeSpecsGauge,
} from './core/metrics';
import { handleResponseTime } from './core/middleware';
import { handleHealthCheck } from './core/middleware';
import { errorHandler, selectMostRecentBySlot, sleep } from './utils/utils';
import { setGlobalDispatcher, Agent } from 'undici';
import { Metrics } from './core/metricsV2';
setGlobalDispatcher(
new Agent({
@@ -51,11 +46,16 @@ console.log('Redis Clients:', REDIS_CLIENTS);
const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
const commitHash = process.env.COMMIT;
const endpoint = process.env.ENDPOINT;
const wsEndpoint = process.env.WS_ENDPOINT;
//@ts-ignore
const sdkConfig = initialize({ env: process.env.ENV });
const stateCommitment: Commitment = 'confirmed';
const serverPort = process.env.PORT || 6969;
const metricsPort = process.env.METRICS_PORT
? parseInt(process.env.METRICS_PORT)
: 9464;
const SLOT_STALENESS_TOLERANCE =
parseInt(process.env.SLOT_STALENESS_TOLERANCE) || 100000;
@@ -65,12 +65,27 @@ const logHttp = morgan(logFormat, {
skip: (_req, res) => res.statusCode <= 500,
});
// init metrics
const metricsV2 = new Metrics('dlob-server-lite', [], metricsPort);
const healthStatusGauge = metricsV2.addGauge(
'health_status',
'Health check status'
);
const cacheHitCounter = metricsV2.addCounter(
'cache_hits',
'Number of cache hits/misses'
);
const incomingRequestsCounter = metricsV2.addCounter(
'incoming_requests',
'Number of incoming requests'
);
metricsV2.finalizeObservables();
const app = express();
app.use(cors({ origin: '*' }));
app.use(compression());
app.set('trust proxy', 1);
app.use(logHttp);
app.use(handleResponseTime());
// strip off /dlob, if the request comes from exchange history server LB
app.use((req, _res, next) => {
@@ -80,19 +95,10 @@ app.use((req, _res, next) => {
req.url = '/';
}
}
incomingRequestsCounter.add(1);
next();
});
// Metrics defined here
const bootTimeMs = Date.now();
runtimeSpecsGauge.addCallback((obs) => {
obs.observe(bootTimeMs, {
commit: commitHash,
driftEnv,
rpcEndpoint: endpoint,
wsEndpoint: wsEndpoint,
incomingRequestsCounter.add(1, {
path: req.baseUrl + req.path,
});
next();
});
app.use(errorHandler);
@@ -108,8 +114,6 @@ server.headersTimeout = 65 * 1000;
const opts = program.opts();
setLogLevel(opts.debug ? 'debug' : 'info');
const endpoint = process.env.ENDPOINT;
const wsEndpoint = process.env.WS_ENDPOINT;
logger.info(`RPC endpoint: ${endpoint}`);
logger.info(`WS endpoint: ${wsEndpoint}`);
logger.info(`DriftEnv: ${driftEnv}`);
@@ -153,9 +157,9 @@ const main = async (): Promise<void> => {
}
};
app.get('/health', handleHealthCheck(slotSubscriber));
app.get('/health', handleHealthCheck(slotSubscriber, healthStatusGauge));
app.get('/startup', handleStartup);
app.get('/', handleHealthCheck(slotSubscriber));
app.get('/', handleHealthCheck(slotSubscriber, healthStatusGauge));
app.get('/l3', async (req, res, next) => {
try {
@@ -178,6 +182,8 @@ const main = async (): Promise<void> => {
cacheHitCounter.add(1, {
miss: false,
path: req.baseUrl + req.path,
marketIndex: normedMarketIndex,
marketType: isSpot ? 'spot' : 'perp',
});
res.writeHead(200);
res.end(JSON.stringify(redisL3));
@@ -186,6 +192,8 @@ const main = async (): Promise<void> => {
cacheHitCounter.add(1, {
miss: true,
path: req.baseUrl + req.path,
marketIndex: normedMarketIndex,
marketType: isSpot ? 'spot' : 'perp',
});
res.writeHead(500);
res.end(JSON.stringify({ error: 'No cached L3 found' }));

10
src/utils/offload.ts
View File

@@ -5,7 +5,7 @@ import {
} from '@aws-sdk/client-kinesis';
import { ConfiguredRetryStrategy } from '@aws-sdk/util-retry';
import { logger } from '../utils/logger';
import { setKinesisRecordsSent } from '../core/metrics';
import { CounterValue } from '../core/metricsV2';
type EventType = 'DLOBSnapshot' | 'DLOBL3Snapshot';
@@ -29,7 +29,7 @@ const batchInterval = process.env.KINESIS_BATCH_INTERVAL
? parseInt(process.env.KINESIS_BATCH_INTERVAL)
: 10000;
export const OffloadQueue = () => {
export const OffloadQueue = (offloadQueueCounter?: CounterValue) => {
const queue: QueueMessage[] = [];
const throttleMap: Map<string, ThrottleInfo> = new Map();
let isProcessing = false;
@@ -84,7 +84,11 @@ export const OffloadQueue = () => {
}
const successCount = batch.length - (response.FailedRecordCount || 0);
setKinesisRecordsSent(successCount, kinesisStream);
if (offloadQueueCounter) {
offloadQueueCounter.add(successCount, {
stream: kinesisStream,
});
}
} catch (error) {
logger.error('Error processing queue:', error);
} finally {

56
src/wsConnectionManager.ts
View File

@@ -4,33 +4,37 @@ import * as http from 'http';
import compression from 'compression';
import { WebSocket, WebSocketServer } from 'ws';
import { sleep, selectMostRecentBySlot, GROUPING_OPTIONS } from './utils/utils';
import { register, Gauge, Counter } from 'prom-client';
import { DriftEnv, PerpMarkets, SpotMarkets } from '@drift-labs/sdk';
import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
import { Metrics } from './core/metricsV2';
// Set up env constants
require('dotenv').config();
const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
const metricsPort = process.env.METRICS_PORT
? parseInt(process.env.METRICS_PORT)
: 9464;
const app = express();
app.use(cors({ origin: '*' }));
app.use(compression());
app.set('trust proxy', 1);
const wsConnectionsGauge = new Gauge({
name: 'websocket_connections',
help: 'Number of active WebSocket connections',
});
const wsOrderbookSourceCounter = new Counter({
name: 'websocket_orderbook_source',
help: 'Number of orderbook messages sent from source',
labelNames: ['source'],
});
const wsOrderbookSourceLastSlotGauge = new Gauge({
name: 'websocket_orderbook_source_last_slot',
help: 'Last slot of orderbook messages from a source',
labelNames: ['source'],
});
// init metrics
const metricsV2 = new Metrics('ws-connection-manager', undefined, metricsPort);
const wsConnectionsCounter = metricsV2.addCounter(
'websocket_connections',
'Number of active WebSocket connections'
);
const wsOrderbookSourceCounter = metricsV2.addCounter(
'websocket_orderbook_source',
'Number of orderbook messages sent from source'
);
const wsOrderbookSourceLastSlotGauge = metricsV2.addGauge(
'websocket_orderbook_source_last_slot',
'Last slot of orderbook messages from a source'
);
metricsV2.finalizeObservables();
const server = http.createServer(app);
const wss = new WebSocketServer({
@@ -143,12 +147,9 @@ async function main() {
if (subscribers) {
if (sanitizedChannel.includes('orderbook')) {
const messageSlot = JSON.parse(message)['slot'];
wsOrderbookSourceLastSlotGauge.set(
{
source: channelPrefix,
},
messageSlot
);
wsOrderbookSourceLastSlotGauge.setLatestValue(messageSlot, {
source: channelPrefix,
});
const lastMessageSlot = subscribedChannelToSlot.get(sanitizedChannel);
if (!lastMessageSlot || lastMessageSlot <= messageSlot) {
@@ -162,7 +163,7 @@ async function main() {
ws.readyState === WebSocket.OPEN &&
ws.bufferedAmount < MAX_BUFFERED_AMOUNT
) {
wsOrderbookSourceCounter.inc({
wsOrderbookSourceCounter.add(1, {
source: channelPrefix,
});
ws.send(
@@ -185,8 +186,7 @@ async function main() {
const subscribedChannels = new Set<string>();
wss.on('connection', (ws: WebSocket) => {
console.log('Client connected');
wsConnectionsGauge.inc();
wsConnectionsCounter.add(1, {});
ws.on('message', async (msg) => {
let parsedMessage: any;
@@ -356,7 +356,6 @@ async function main() {
// Handle disconnection
ws.on('close', () => {
console.log('Client disconnected');
// Clear any existing intervals
clearInterval(heartbeatInterval);
clearInterval(bufferInterval);
@@ -367,7 +366,7 @@ async function main() {
subscribedChannels.delete(channel);
}
});
wsConnectionsGauge.dec();
wsConnectionsCounter.add(-1, {});
});
ws.on('error', (error) => {
@@ -379,11 +378,6 @@ async function main() {
console.log(`connection manager running on ${WS_PORT}`);
});
app.get('/metrics', async (req, res) => {
res.set('Content-Type', register.contentType);
res.end(await register.metrics());
});
server.on('error', (error) => {
console.error('Server error:', error);
});

4179
yarn.lock
View File

File diff suppressed because it is too large Load Diff