Merge pull request #455 from drift-labs/wphan/health_check_fix

refactor health checks and prettify/lint

package.json

@@ -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/eslint-plugin": "6.21.0",
-		"@typescript-eslint/parser": "^4.28.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"
 	}
 }

src/core/healthCheck.ts

@@ -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,
+};

src/core/metrics.ts

@@ -1,253 +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',
-}
-
-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 healthCheckInterval = 2000;
-let lastHealthCheckSlot = -1;
-let lastHealthCheckState = true; // true = healthy, false = unhealthy
-let lastHealthCheckPerformed = Date.now() - healthCheckInterval;
-let lastTimeHealthy = Date.now() - healthCheckInterval;
-
-export const setHealthStatus = (status: HEALTH_STATUS): void => {
-	healthStatus = status;
-};
-
-/**
- * Middleware that checks if we are in general healthy by checking that the bulk account loader slot
- * has changed recently.
- *
- * We may be hit by multiple sources performing health checks on us, so this middleware will latch
- * to its health state and only update every `healthCheckInterval`.
- *
- * A grace period is also used to report unhealthy only if we have been unhealthy beyond the grace period.
- */
-const handleHealthCheck = (
-	healthCheckGracePeriod: number,
-	slotSource: SlotSource
-) => {
-	return async (_req, res, _next) => {
-		if (healthStatus === HEALTH_STATUS.Restart) {
-			logger.error(`Health status: Restart`);
-			res.writeHead(500);
-			res.end(`NOK`);
-			return;
-		}
-
-		if (Date.now() < lastHealthCheckPerformed + healthCheckInterval) {
-			if (lastHealthCheckState) {
-				res.writeHead(200);
-				res.end('OK');
-				lastHealthCheckPerformed = Date.now();
-				return;
-			}
-			// always check if last check was unhealthy (give it another chance to recover)
-		}
-
-		// healthy if slot has advanced since the last check
-		const lastSlotReceived = slotSource.getSlot();
-		const inGracePeriod =
-			Date.now() - lastTimeHealthy <= healthCheckGracePeriod;
-		lastHealthCheckState = lastSlotReceived > lastHealthCheckSlot;
-		if (!lastHealthCheckState) {
-			logger.error(
-				`Unhealthy: lastSlot: ${lastSlotReceived}, lastHealthCheckSlot: ${lastHealthCheckSlot}, timeSinceLastCheck: ${
-					Date.now() - lastHealthCheckPerformed
-				} ms, sinceLastTimeHealthy: ${
-					Date.now() - lastTimeHealthy
-				} ms, inGracePeriod: ${inGracePeriod}`
-			);
-		} else {
-			lastTimeHealthy = Date.now();
-		}
-
-		lastHealthCheckSlot = lastSlotReceived;
-		lastHealthCheckPerformed = Date.now();
-
-		if (!lastHealthCheckState && !inGracePeriod) {
-			setHealthStatus(HEALTH_STATUS.UnhealthySlotSubscriber);
-
-			res.writeHead(500);
-			res.end(`NOK`);
-			return;
-		}
-
-		setHealthStatus(HEALTH_STATUS.Ok);
-		res.writeHead(200);
-		res.end('OK');
-	};
-};
-
-export {
-	endpointResponseTimeHistogram,
-	gpaFetchDurationHistogram,
-	responseStatusCounter,
-	incomingRequestsCounter,
-	handleHealthCheck,
-	setLastReceivedWsMsgTs,
-	accountUpdatesCounter,
-	cacheHitCounter,
-	runtimeSpecsGauge,
-};

src/core/metricsV2.ts

@@ -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())
+			);
+		}
+	}
+}

src/core/middleware.ts

@@ -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, {
+		// TODO: add these back if want to profile response times
-			endpoint,
-			status: res.statusCode,
-		});
 
-		const responseTimeMs = time;
+		// responseStatusCounter.add(1, {
-		endpointResponseTimeHistogram.record(responseTimeMs, {
+		// 	endpoint,
-			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');
+	};
+};

src/dlob-subscriber/DLOBSubscriberIO.ts

@@ -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

src/index.ts

@@ -28,15 +28,8 @@ import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
 import { logger, setLogLevel } from './utils/logger';
 
 import * as http from 'http';
-import {
+import { Metrics } from './core/metricsV2';
-	handleHealthCheck,
+import { handleHealthCheck } from './core/middleware';
-	accountUpdatesCounter,
-	cacheHitCounter,
-	setLastReceivedWsMsgTs,
-	incomingRequestsCounter,
-	runtimeSpecsGauge,
-} from './core/metrics';
-import { handleResponseTime } 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());
+			lastWsReceivedTsGauge.setLatestValue(Date.now(), {});
-			// eslint-disable-next-line @typescript-eslint/no-unused-vars
+			accountUpdatesCounter.add(1, {});
-			updatesReceivedTotal++;
-			accountUpdatesCounter.add(1);
 		}
 	);
 
@@ -265,12 +271,9 @@ const main = async (): Promise<void> => {
 		}
 	};
 
-	app.get(
+	app.get('/health', handleHealthCheck(dlobProvider, healthStatusGauge));
-		'/health',
-		handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, dlobProvider)
-	);
 	app.get('/startup', handleStartup);
-	app.get('/', handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, dlobProvider));
+	app.get('/', handleHealthCheck(dlobProvider, healthStatusGauge));
 
 	app.get('/priorityFees', async (req, res, next) => {
 		try {

src/publishers/dlobPublisher.ts

@@ -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
+// init metrics
-const dlobSlotGauge = new Gauge({
+const metricsV2 = new Metrics('dlob-publisher', undefined, metricsPort);
-	name: 'dlob_slot',
+const healthStatusGauge = metricsV2.addGauge(
-	help: 'Last updated slot of DLOB',
+	'health_status',
-	labelNames: [
+	'Health check status'
-		'marketIndex',
+);
-		'marketType',
+const dlobSlotGauge = metricsV2.addGauge(
-		'marketName',
+	'dlob_slot',
-		'redisPrefix',
+	'Last updated slot of DLOB'
-		'redisClient',
+);
-	],
+const oracleSlotGauge = metricsV2.addGauge(
-});
+	'oracle_slot',
-const oracleSlotGauge = new Gauge({
+	'Last updated slot of oracle'
-	name: 'oracle_slot',
+);
-	help: 'Last updated slot of oracle',
+const kinesisRecordsSentCounter = metricsV2.addCounter(
-	labelNames: [
+	'kinesis_records_sent',
-		'marketIndex',
+	'Number of records sent to Kinesis'
-		'marketType',
+);
-		'marketName',
+metricsV2.finalizeObservables();
-		'redisPrefix',
-		'redisClient',
-	],
-});
 
 //@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(
+			dlobSlotGauge.setLatestValue(slot, {
-				{
+				marketIndex: market.marketIndex,
-					marketIndex: market.marketIndex,
+				marketType: 'perp',
-					marketType: 'perp',
+				marketName: market.marketName,
-					marketName: market.marketName,
+				redisClient: REDIS_CLIENT,
-					redisClient: REDIS_CLIENT,
+				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
-					redisPrefix: RedisClientPrefix[REDIS_CLIENT],
+			});
-				},
+			oracleSlotGauge.setLatestValue(oracleDataAndSlot.slot.toNumber(), {
-				slot
+				marketIndex: market.marketIndex,
-			);
+				marketType: 'perp',
-			oracleSlotGauge.set(
+				marketName: market.marketName,
-				{
+				redisClient: REDIS_CLIENT,
-					marketIndex: market.marketIndex,
+				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
-					marketType: 'perp',
+			});
-					marketName: market.marketName,
-					redisClient: REDIS_CLIENT,
-					redisPrefix: RedisClientPrefix[REDIS_CLIENT],
-				},
-				oracleDataAndSlot.slot.toNumber()
-			);
 		});
 		spotMarketInfos.forEach((market) => {
 			const oracleDataAndSlot = driftClient.getOracleDataForSpotMarket(
 				market.marketIndex
 			);
-			dlobSlotGauge.set(
+			dlobSlotGauge.setLatestValue(oracleDataAndSlot.slot.toNumber(), {
-				{
+				marketIndex: market.marketIndex,
-					marketIndex: market.marketIndex,
+				marketType: 'spot',
-					marketType: 'spot',
+				marketName: market.marketName,
-					marketName: market.marketName,
+				redisClient: REDIS_CLIENT,
-					redisClient: REDIS_CLIENT,
+				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
-					redisPrefix: RedisClientPrefix[REDIS_CLIENT],
+			});
-				},
-				oracleDataAndSlot.slot.toNumber()
-			);
 		});
 	}, 10_000);
 
@@ -646,18 +637,9 @@ const main = async () => {
 		}
 	};
 	app.get('/debug', handleDebug);
-
+	app.get('/health', handleHealthCheck(slotSource, healthStatusGauge));
-	app.get('/metrics', async (req, res) => {
-		res.set('Content-Type', register.contentType);
-		res.end(await register.metrics());
-	});
-
-	app.get(
-		'/health',
-		handleHealthCheck(WS_FALLBACK_FETCH_INTERVAL, dlobProvider)
-	);
 	app.get('/startup', handleStartup);
-	app.get('/', handleHealthCheck(WS_FALLBACK_FETCH_INTERVAL, dlobProvider));
+	app.get('/', handleHealthCheck(slotSource, healthStatusGauge));
 	const server = app.listen(8080);
 
 	// Default keepalive is 5s, since the AWS ALB timeout is 60 seconds, clients

src/publishers/priorityFeesPublisher.ts

@@ -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(

src/serverLite.ts

@@ -18,15 +18,10 @@ import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
 import { logger, setLogLevel } from './utils/logger';
 
 import * as http from 'http';
-import {
+import { handleHealthCheck } from './core/middleware';
-	handleHealthCheck,
-	cacheHitCounter,
-	incomingRequestsCounter,
-	runtimeSpecsGauge,
-} from './core/metrics';
-import { handleResponseTime } from './core/middleware';
 import { errorHandler, selectMostRecentBySlot, sleep } from './utils/utils';
 import { setGlobalDispatcher, Agent } from 'undici';
+import { Metrics } from './core/metricsV2';
 
 setGlobalDispatcher(
 	new Agent({
@@ -51,13 +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;
-export const ORDERBOOK_UPDATE_INTERVAL = 1000;
+const metricsPort = process.env.METRICS_PORT
-const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 60;
+	? parseInt(process.env.METRICS_PORT)
+	: 9464;
 const SLOT_STALENESS_TOLERANCE =
 	parseInt(process.env.SLOT_STALENESS_TOLERANCE) || 100000;
 
@@ -67,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) => {
@@ -82,19 +95,10 @@ app.use((req, _res, next) => {
 			req.url = '/';
 		}
 	}
-	incomingRequestsCounter.add(1);
+	incomingRequestsCounter.add(1, {
-	next();
+		path: req.baseUrl + req.path,
-});
-
-// Metrics defined here
-const bootTimeMs = Date.now();
-runtimeSpecsGauge.addCallback((obs) => {
-	obs.observe(bootTimeMs, {
-		commit: commitHash,
-		driftEnv,
-		rpcEndpoint: endpoint,
-		wsEndpoint: wsEndpoint,
 	});
+	next();
 });
 
 app.use(errorHandler);
@@ -110,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}`);
@@ -155,15 +157,9 @@ const main = async (): Promise<void> => {
 		}
 	};
 
-	app.get(
+	app.get('/health', handleHealthCheck(slotSubscriber, healthStatusGauge));
-		'/health',
-		handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, slotSubscriber)
-	);
 	app.get('/startup', handleStartup);
-	app.get(
+	app.get('/', handleHealthCheck(slotSubscriber, healthStatusGauge));
-		'/',
-		handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, slotSubscriber)
-	);
 
 	app.get('/l3', async (req, res, next) => {
 		try {
@@ -186,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));
@@ -194,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' }));

src/utils/constants.ts

@@ -1,4 +1,4 @@
-import { AuctionParamArgs } from "./types";
+import { AuctionParamArgs } from './types';
 
 export const MEASURED_ENDPOINTS = [
 	'/priorityFees',

src/utils/offload.ts

@@ -5,6 +5,7 @@ import {
 } from '@aws-sdk/client-kinesis';
 import { ConfiguredRetryStrategy } from '@aws-sdk/util-retry';
 import { logger } from '../utils/logger';
+import { CounterValue } from '../core/metricsV2';
 
 type EventType = 'DLOBSnapshot' | 'DLOBL3Snapshot';
 
@@ -28,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;
@@ -83,9 +84,11 @@ export const OffloadQueue = () => {
 			}
 
 			const successCount = batch.length - (response.FailedRecordCount || 0);
-			logger.info(
+			if (offloadQueueCounter) {
-				`Successfully sent ${successCount} records to Kinesis stream: ${kinesisStream}`
+				offloadQueueCounter.add(successCount, {
-			);
+					stream: kinesisStream,
+				});
+			}
 		} catch (error) {
 			logger.error('Error processing queue:', error);
 		} finally {

src/utils/tests/auctionParams.test.ts

@@ -17,7 +17,7 @@ import {
 describe('Auction Parameters Functions', () => {
 	describe('createMarketBasedAuctionParams', () => {
 		it('should apply market-based logic for major PERP markets (0, 1, 2)', () => {
-			[0, 1, 2].forEach(marketIndex => {
+			[0, 1, 2].forEach((marketIndex) => {
 				const args = {
 					marketIndex,
 					marketType: 'perp' as any,
@@ -36,7 +36,7 @@ describe('Auction Parameters Functions', () => {
 		});
 
 		it('should apply market-based logic for minor PERP markets (>2)', () => {
-			[3, 4, 5, 10].forEach(marketIndex => {
+			[3, 4, 5, 10].forEach((marketIndex) => {
 				const args = {
 					marketIndex,
 					marketType: 'perp' as any,
@@ -55,7 +55,7 @@ describe('Auction Parameters Functions', () => {
 		});
 
 		it('should apply market-based logic for SPOT markets when "marketBased" is passed', () => {
-			[0, 1, 2, 5, 10].forEach(marketIndex => {
+			[0, 1, 2, 5, 10].forEach((marketIndex) => {
 				const args = {
 					marketIndex,
 					marketType: 'spot' as any,
@@ -382,7 +382,9 @@ describe('Auction Parameters Functions', () => {
 			);
 
 			expect(result.success).toBe(true);
-			expect(result.data?.marketOrderParams.direction).toBe(PositionDirection.SHORT);
+			expect(result.data?.marketOrderParams.direction).toBe(
+				PositionDirection.SHORT
+			);
 			expect(result.data?.estimatedPrices.entryPrice.gt(ZERO)).toBe(true);
 		});
 
@@ -398,12 +400,12 @@ describe('Auction Parameters Functions', () => {
 				if (key.includes('dlob_orderbook')) {
 					return {
 						bids: [
-							{ price: '159950000', size: '500000000' },  // $159.95, 0.5 SOL
+							{ price: '159950000', size: '500000000' }, // $159.95, 0.5 SOL
 							{ price: '159900000', size: '1000000000' }, // $159.90, 1 SOL
 							{ price: '159850000', size: '2000000000' }, // $159.85, 2 SOL
 						],
 						asks: [
-							{ price: '160050000', size: '500000000' },  // $160.05, 0.5 SOL
+							{ price: '160050000', size: '500000000' }, // $160.05, 0.5 SOL
 							{ price: '160100000', size: '1000000000' }, // $160.10, 1 SOL
 							{ price: '160150000', size: '2000000000' }, // $160.15, 2 SOL
 						],
@@ -445,7 +447,11 @@ describe('Auction Parameters Functions', () => {
 			expect(largeResult.data?.estimatedPrices.entryPrice.gt(ZERO)).toBe(true);
 
 			// Large orders should have higher price impact
-			expect(largeResult.data?.estimatedPrices.priceImpact.gte(smallResult.data?.estimatedPrices.priceImpact)).toBe(true);
+			expect(
+				largeResult.data?.estimatedPrices.priceImpact.gte(
+					smallResult.data?.estimatedPrices.priceImpact
+				)
+			).toBe(true);
 		});
 
 		it('should handle zero oracle price scenario', async () => {
@@ -601,11 +607,15 @@ describe('Auction Parameters Functions', () => {
 				},
 			];
 
-			scenarios.forEach(scenario => {
+			scenarios.forEach((scenario) => {
 				const result = createMarketBasedAuctionParams(scenario.input);
 
-				expect(result.auctionStartPriceOffsetFrom).toBe(scenario.expectedOffset.from);
+				expect(result.auctionStartPriceOffsetFrom).toBe(
-				expect(result.auctionStartPriceOffset).toBe(scenario.expectedOffset.value);
+					scenario.expectedOffset.from
+				);
+				expect(result.auctionStartPriceOffset).toBe(
+					scenario.expectedOffset.value
+				);
 
 				// Verify all original parameters are preserved
 				expect(result.marketIndex).toBe(scenario.input.marketIndex);
@@ -667,7 +677,8 @@ describe('Auction Parameters Functions', () => {
 				isOracleOrder: true,
 			};
 
-			const formattedResponse = formatAuctionParamsForResponse(mockDerivedParams);
+			const formattedResponse =
+				formatAuctionParamsForResponse(mockDerivedParams);
 
 			// Verify the complete flow
 			expect(formattedResponse.auctionStartPriceOffsetFrom).toBe('mark'); // Market-based for major market

src/utils/types.ts

@@ -1,5 +1,5 @@
-import { AssetType, MarketTypeStr } from "@drift-labs/sdk";
+import { AssetType, MarketTypeStr } from '@drift-labs/sdk';
-import { TradeOffsetPrice } from "@drift/common";
+import { TradeOffsetPrice } from '@drift/common';
 
 export type AuctionParamArgs = {
 	// mandatory args

src/utils/utils.ts

@@ -656,14 +656,16 @@ export function createMarketBasedAuctionParams(
 
 	// Resolve "marketBased" values and undefined values (both should use market-based logic)
 	const resolvedAuctionStartPriceOffsetFrom =
-		args.auctionStartPriceOffsetFrom === 'marketBased' || args.auctionStartPriceOffsetFrom === undefined
+		args.auctionStartPriceOffsetFrom === 'marketBased' ||
+		args.auctionStartPriceOffsetFrom === undefined
 			? isMajorMarket
 				? 'mark'
 				: 'bestOffer'
 			: args.auctionStartPriceOffsetFrom;
 
 	const resolvedAuctionStartPriceOffset =
-		args.auctionStartPriceOffset === 'marketBased' || args.auctionStartPriceOffset === undefined
+		args.auctionStartPriceOffset === 'marketBased' ||
+		args.auctionStartPriceOffset === undefined
 			? isMajorMarket
 				? 0
 				: -0.1

src/wsConnectionManager.ts

@@ -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({
+// init metrics
-	name: 'websocket_connections',
+const metricsV2 = new Metrics('ws-connection-manager', undefined, metricsPort);
-	help: 'Number of active WebSocket connections',
+const wsConnectionsCounter = metricsV2.addCounter(
-});
+	'websocket_connections',
-const wsOrderbookSourceCounter = new Counter({
+	'Number of active WebSocket connections'
-	name: 'websocket_orderbook_source',
+);
-	help: 'Number of orderbook messages sent from source',
+const wsOrderbookSourceCounter = metricsV2.addCounter(
-	labelNames: ['source'],
+	'websocket_orderbook_source',
-});
+	'Number of orderbook messages sent from source'
-const wsOrderbookSourceLastSlotGauge = new Gauge({
+);
-	name: 'websocket_orderbook_source_last_slot',
+const wsOrderbookSourceLastSlotGauge = metricsV2.addGauge(
-	help: 'Last slot of orderbook messages from a source',
+	'websocket_orderbook_source_last_slot',
-	labelNames: ['source'],
+	'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(
+					wsOrderbookSourceLastSlotGauge.setLatestValue(messageSlot, {
-						{
+						source: channelPrefix,
-							source: channelPrefix,
+					});
-						},
-						messageSlot
-					);
 
 					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');
+		wsConnectionsCounter.add(1, {});
-		wsConnectionsGauge.inc();
 
 		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);
 	});