Merge pull request #516 from drift-labs/wphan/pyth-cranker-metrics

pyth-crankers: add tx recorder metrics and health check

Author: wphan

src/bots/common/txRecorder.ts

@@ -6,16 +6,28 @@ import {
 	ExplicitBucketHistogramAggregation,
 	InstrumentType,
 } from '@opentelemetry/sdk-metrics-base';
-import { metrics, type Histogram } from '@opentelemetry/api';
+import { metrics, ObservableGauge, type Histogram } from '@opentelemetry/api';
 import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
 import { registerInstrumentations } from '@opentelemetry/instrumentation';
 
 export class TxRecorder {
 	private txLatency?: Histogram;
 	private slotLatency?: Histogram;
+	private txLatencyEma?: ObservableGauge;
 	private attrs: Record<string, string> = {};
+	private movingAvgLatencyMs?: number;
+	private emaAlpha: number = 0.1;
+	private healthThresholdMs: number = 20000;
+
+	constructor(
+		botName: string,
+		port?: number,
+		disabled?: boolean,
+		thresholdMs = 20000
+	) {
+		// set health threshold regardless of metrics being enabled
+		this.healthThresholdMs = thresholdMs;
 
-	constructor(botName: string, port?: number, disabled?: boolean) {
 		if (disabled || !port) {
 			logger.info(`metrics disabled for bot ${botName}`);
 			return;
@@ -71,10 +83,30 @@ export class TxRecorder {
 		this.slotLatency = meter.createHistogram('tx_slot_latency', {
 			unit: 'slots',
 		});
+		this.txLatencyEma = meter.createObservableGauge('tx_latency_ema', {
+			unit: 'ms',
+		});
+		this.txLatencyEma.addCallback((result) => {
+			result.observe(this.movingAvgLatencyMs ?? 0, this.attrs);
+		});
 	}
 
 	send(latencyMs: number, slotDelta = 0) {
 		this.txLatency?.record(latencyMs, this.attrs);
 		this.slotLatency?.record(Math.max(0, slotDelta), this.attrs);
+
+		// update exponential moving average of tx latency
+		if (this.movingAvgLatencyMs === undefined) {
+			this.movingAvgLatencyMs = latencyMs;
+		} else {
+			this.movingAvgLatencyMs =
+				this.emaAlpha * latencyMs +
+				(1 - this.emaAlpha) * this.movingAvgLatencyMs;
+		}
+	}
+
+	isHealthy(): boolean {
+		if (this.movingAvgLatencyMs === undefined) return true;
+		return this.movingAvgLatencyMs <= this.healthThresholdMs;
 	}
 }

src/bots/pythCranker.ts

@@ -130,7 +130,12 @@ export class PythCrankerBot implements Bot {
 		this.slotStalenessThresholdRestart =
 			crankConfigs.slotStalenessThresholdRestart;
 
-		this.txRecorder = new TxRecorder(this.name, crankConfigs.metricsPort);
+		this.txRecorder = new TxRecorder(
+			this.name,
+			crankConfigs.metricsPort,
+			false,
+			20_000
+		);
 	}
 
 	async init(): Promise<void> {
@@ -487,16 +492,14 @@ export class PythCrankerBot implements Bot {
 						this.driftClient
 							.sendTransaction(simResult.tx)
 							.then((txSigAndSlot: TxSigAndSlot) => {
-								this.txRecorder.send(
-									Date.now() - startTime,
-									txSigAndSlot.slot - sendSlot
-								);
+								const duration = Date.now() - startTime;
+								this.txRecorder.send(duration, txSigAndSlot.slot - sendSlot);
 								logger.info(
 									`Posted multi pyth pull oracle for ${feedIds.map(
 										(feedId) => feedId.baseSymbol
-									)} update atomic tx: ${txSigAndSlot.txSig}, took ${
-										Date.now() - startTime
-									}ms, landed slot: ${
+									)} update atomic tx: ${
+										txSigAndSlot.txSig
+									}, took ${duration}ms, landed slot: ${
 										txSigAndSlot.slot
 									}, sent slot: ${sendSlot}`
 								);
@@ -534,6 +537,11 @@ export class PythCrankerBot implements Bot {
 	}
 
 	async healthCheck(): Promise<boolean> {
+		const txRecorderHealthy = this.txRecorder.isHealthy();
+		if (!txRecorderHealthy) {
+			logger.warn(`${this.name} bot tx recorder is unhealthy`);
+		}
+		this.health = txRecorderHealthy;
 		return this.health;
 	}
 }

src/bots/pythLazerCranker.ts

@@ -30,6 +30,7 @@ import {
 	PythLazerSubscriber,
 } from '../pythLazerSubscriber';
 import { Channel } from '@pythnetwork/pyth-lazer-sdk';
+import { TxRecorder } from './common/txRecorder';
 
 setGlobalDispatcher(
 	new Agent({
@@ -50,6 +51,8 @@ export class PythLazerCrankerBot implements Bot {
 
 	private blockhashSubscriber: BlockhashSubscriber;
 	private health: boolean = true;
+	// Metrics
+	private txRecorder: TxRecorder;
 
 	constructor(
 		private globalConfig: GlobalConfig,
@@ -150,6 +153,13 @@ export class PythLazerCrankerBot implements Bot {
 		this.blockhashSubscriber = new BlockhashSubscriber({
 			connection: driftClient.connection,
 		});
+
+		this.txRecorder = new TxRecorder(
+			this.name,
+			crankConfigs.metricsPort,
+			false,
+			20_000
+		);
 	}
 
 	async init(): Promise<void> {
@@ -257,10 +267,10 @@ export class PythLazerCrankerBot implements Bot {
 				this.driftClient
 					.sendTransaction(simResult.tx)
 					.then((txSigAndSlot: TxSigAndSlot) => {
+						const duration = Date.now() - startTime;
+						this.txRecorder.send(duration);
 						logger.info(
-							`Posted pyth lazer oracles for ${feedIds} update atomic tx: ${
-								txSigAndSlot.txSig
-							}, took ${Date.now() - startTime}ms`
+							`Posted pyth lazer oracles for ${feedIds} update atomic tx: ${txSigAndSlot.txSig}, took ${duration}ms, skippedSim: false`
 						);
 					})
 					.catch((e) => {
@@ -277,10 +287,10 @@ export class PythLazerCrankerBot implements Bot {
 				this.driftClient
 					.sendTransaction(tx)
 					.then((txSigAndSlot: TxSigAndSlot) => {
+						const duration = Date.now() - startTime;
+						this.txRecorder.send(duration);
 						logger.info(
-							`Posted pyth lazer oracles for ${feedIds} update atomic tx: ${
-								txSigAndSlot.txSig
-							}, took ${Date.now() - startTime}ms`
+							`Posted pyth lazer oracles for ${feedIds} update atomic tx: ${txSigAndSlot.txSig}, took ${duration}ms, skippedSim: true`
 						);
 					})
 					.catch((e) => {
@@ -291,6 +301,11 @@ export class PythLazerCrankerBot implements Bot {
 	}
 
 	async healthCheck(): Promise<boolean> {
+		const txRecorderHealthy = this.txRecorder.isHealthy();
+		if (!txRecorderHealthy) {
+			logger.warn(`${this.name} bot tx recorder is unhealthy`);
+		}
+		this.health = txRecorderHealthy;
 		return this.health;
 	}
 }