Merge pull request #166 from D8-X/track_rpc

Track rpc

packages/api/src/eventListener.ts

@@ -39,6 +39,7 @@ import {
 } from "utils/src/wsTypes";
 import SturdyWebSocket from "sturdy-websocket";
 import { Logger } from "winston";
+import { TrackedWebsocketsProvider } from "./providers";
 
 /**
  * Class that listens to blockchain events on
@@ -201,7 +202,7 @@ export default class EventListener extends IndexPriceInterface {
 
 		// Attempt to establish a ws connection to new RPC
 		this.logger.info("creating new websocket rpc provider");
-		this.currentWSRpcProvider = new providers.WebSocketProvider(this.wsConn!);
+		this.currentWSRpcProvider = new TrackedWebsocketsProvider(this.wsConn!);
 
 		// On provider error - retry after short cooldown
 		this.currentWSRpcProvider.on("error", (error: Error) => () => {
@@ -632,7 +633,7 @@ export default class EventListener extends IndexPriceInterface {
 	 * @param symbol order book symbol
 	 */
 	private addOrderBookEventHandlers(symbol: string) {
-		const provider = new providers.WebSocketProvider(this.wsRPC);
+		const provider = this.currentWSRpcProvider!;
 		this.orderBookContracts[symbol] = new Contract(
 			this.traderInterface.getOrderBookAddress(symbol),
 			this.traderInterface.getABI("lob")!,

packages/api/src/index.ts

@@ -9,6 +9,13 @@ import * as winston from "winston";
 import { RPCConfig } from "utils/dist/wsTypes";
 import RPCManager from "./rpcManager";
 import fs from "fs";
+import {
+	JsonRpcEthCalls,
+	NumJsonRpcProviders,
+	NumWssProviders,
+	ProvidersEthCallsStartTime,
+	WssEthCalls,
+} from "./providers";
 
 const defaultLogger = () => {
 	return winston.createLogger({
@@ -112,6 +119,22 @@ async function start() {
 		} else {
 			waitTime = 60_000;
 		}
+
+		// Print out eth calls statistics
+		const currentTime = new Date();
+		console.log("statistics of eth_ calls", {
+			JsonRpcEthCalls: JsonRpcEthCalls,
+			WssEthCalls: WssEthCalls,
+			CurrentTime: currentTime.toISOString(),
+			StartTime: ProvidersEthCallsStartTime.toISOString(),
+			RunningFor:
+				(currentTime.getTime() - ProvidersEthCallsStartTime.getTime()) /
+					1000 /
+					60 +
+				" minutes",
+			NumJsonRpcProviders,
+			NumWssProviders,
+		});
 	}
 }
 start();

packages/api/src/providers.ts

@@ -0,0 +1,51 @@
+import { Networkish } from "@ethersproject/providers";
+import { WebSocketLike } from "@ethersproject/providers/lib/websocket-provider";
+import { providers } from "ethers";
+import { ConnectionInfo } from "ethers/lib/utils";
+
+export const ProvidersEthCallsStartTime = new Date();
+
+/**
+ * List of eth_ * method calls and their counts for all TrackedJsonRpcProvider
+ */
+export const JsonRpcEthCalls = new Map<string, number>();
+
+/**
+ * List of eth_ * method calls and their counts for all TrackedWebsocketsProvider
+ */
+export const WssEthCalls = new Map<string, number>();
+
+export let NumJsonRpcProviders = 0;
+export let NumWssProviders = 0;
+
+export class TrackedWebsocketsProvider extends providers.WebSocketProvider {
+	constructor(url: string | WebSocketLike, network?: Networkish) {
+		super(url, network);
+		NumWssProviders++;
+	}
+
+	send(method: string, params?: Array<any>): Promise<any> {
+		if (!WssEthCalls.has(method)) {
+			WssEthCalls.set(method, 0);
+		}
+		WssEthCalls.set(method, WssEthCalls.get(method)! + 1);
+
+		return super.send(method, params);
+	}
+}
+
+export class TrackedJsonRpcProvider extends providers.StaticJsonRpcProvider {
+	constructor(url?: ConnectionInfo | string, network?: Networkish) {
+		super(url, network);
+		NumJsonRpcProviders++;
+	}
+
+	send(method: string, params: Array<any>): Promise<any> {
+		if (!JsonRpcEthCalls.has(method)) {
+			JsonRpcEthCalls.set(method, 0);
+		}
+		JsonRpcEthCalls.set(method, JsonRpcEthCalls.get(method)! + 1);
+
+		return super.send(method, params);
+	}
+}

packages/api/src/rpcManager.ts

@@ -1,5 +1,6 @@
 import { JsonRpcProvider } from "@ethersproject/providers";
 import { executeWithTimeout } from "utils";
+import { TrackedJsonRpcProvider } from "./providers";
 
 export default class RPCManager {
 	private healthy: Map<string, boolean> = new Map();
@@ -54,7 +55,7 @@ export default class RPCManager {
 			this.healthy.get(rpc) === undefined ||
 			(this.lastCheck.get(rpc) ?? 0) + this.CHECK_INTERVAL_MS < Date.now()
 		) {
-			const provider = new JsonRpcProvider(rpc);
+			const provider = new TrackedJsonRpcProvider(rpc);
 			try {
 				await executeWithTimeout(provider.detectNetwork(), this.NETWORK_READY_MS);
 				this.healthy.set(rpc, true);

packages/api/src/sdkInterface.ts

@@ -19,6 +19,7 @@ import Observable from "./observable";
 import type { RedisClientType } from "redis";
 import { extractErrorMsg, constructRedis } from "utils";
 import RPCManager from "./rpcManager";
+import { TrackedJsonRpcProvider } from "./providers";
 
 export type OrderWithTraderAndId = Order & { orderId: string; trader: string };
 
@@ -48,7 +49,9 @@ export default class SDKInterface extends Observable {
 	public async initialize(sdkConfig: NodeSDKConfig, rpcManager: RPCManager) {
 		this.apiInterface = new TraderInterface(sdkConfig);
 		this.rpcManager = rpcManager;
-		await this.apiInterface.createProxyInstance();
+		await this.apiInterface.createProxyInstance(
+			new TrackedJsonRpcProvider(sdkConfig.nodeURL),
+		);
 		await this.broker.initialize(sdkConfig);
 		const brokerAddress = await this.broker.getBrokerAddress();
 		if (!this.redisClient.isOpen) {