feat: agent service healthchecks

Author: kuco23

packages/fasset-bots-cli/src/run/run-agent.ts

@@ -1,12 +1,13 @@
 import "dotenv/config";
 import "source-map-support/register";
 
-import { AgentBotRunner, PricePublisherService, startActivityTimestampUpdater, stopActivityTimestampUpdater, TimeKeeperService, TimekeeperTimingConfig } from "@flarenetwork/fasset-bots-core";
+import { AgentBotRunner, PricePublisherService, startActivityTimestampUpdater, startHealthcheckMonitor, stopActivityTimestampUpdater, TimeKeeperService, TimekeeperTimingConfig } from "@flarenetwork/fasset-bots-core";
 import { closeBotConfig, createBotConfig, loadAgentConfigFile, Secrets } from "@flarenetwork/fasset-bots-core/config";
 import { assertCmd, assertNotNullCmd, authenticatedHttpProvider, CommandLineError, formatFixed, initWeb3, isNotNull, logger, sendWeb3Transaction, toBN, toBNExp, web3 } from "@flarenetwork/fasset-bots-core/utils";
 import BN from "bn.js";
 import { programWithCommonOptions } from "../utils/program";
 import { toplevelRun } from "../utils/toplevel";
+import { AgentNotifier } from "../../../fasset-bots-core/src/utils/notifier/AgentNotifier";
 
 const timekeeperConfig: TimekeeperTimingConfig = {
     queryWindow: 172800,
@@ -17,6 +18,7 @@ const timekeeperConfig: TimekeeperTimingConfig = {
 }
 
 const activityUpdateInterval = 60000; // 1min
+const healthcheckMonitorInterval = 60000; // 1min
 
 const program = programWithCommonOptions("agent", "all_fassets");
 
@@ -104,6 +106,9 @@ program.action(async () => {
         }
         // start activity update
         startActivityTimestampUpdater(botConfig.orm.em, activityUpdateInterval);
+        // start healthcheck monitor
+        const agent = new AgentNotifier(owner.address, runner.notifierTransports);
+        startHealthcheckMonitor(botConfig, agent, healthcheckMonitorInterval);
         // run
         try {
             console.log("Agent bot started, press CTRL+C to end");

packages/fasset-bots-core/src/actors/service-health-checker.ts

@@ -0,0 +1,113 @@
+import { web3 } from "@flarenetwork/fasset-bots-core/utils";
+import { AgentNotifier } from "../utils/notifier/AgentNotifier";
+import { BlockchainIndexerHelper } from "../underlying-chain/BlockchainIndexerHelper";
+import { FlareDataConnectorClientHelper } from "../underlying-chain/FlareDataConnectorClientHelper";
+import { IBlockChainWallet } from "../underlying-chain/interfaces/IBlockChainWallet";
+import { AgentBotConfig } from "../config";
+import { logger } from "../utils/logger";
+
+const MAX_BLOCK_TIMESTAMP_DELAY_SECONDS = 30
+
+async function checkNativeRpcHealth(): Promise<string | undefined> {
+    try {
+        const height = await web3.eth.getBlockNumber()
+        const block = await web3.eth.getBlock(height)
+        const now = Math.floor(Date.now() / 1000)
+        if (Number(block.timestamp) < now - MAX_BLOCK_TIMESTAMP_DELAY_SECONDS) {
+            return 'block height too far behind'
+        }
+    } catch (e: any) {
+        return String(e?.message)
+    }
+}
+
+async function checkIndexerHealth(
+    indexer: BlockchainIndexerHelper
+): Promise<string | undefined> {
+    try {
+        const ok = await indexer.getHealth()
+        if (!ok) {
+            return 'health check failed'
+        }
+    } catch (e: any) {
+        return String(e?.message)
+    }
+}
+
+async function checkDalHealth(
+    fdc: FlareDataConnectorClientHelper
+): Promise<string | undefined> {
+    try {
+        const healthy = await fdc.getHealth()
+        const n = healthy.filter(h => h).length
+        if (n < fdc.dataAccessLayerClients.length) {
+            const unhealthy = healthy.map((h, i) => h ? '' : fdc.dataAccessLayerUrls[i]).join(',')
+            return `clients ${unhealthy} are healthy`
+        } else if (n == 0) {
+            return 'no clients available'
+        }
+    } catch (e: any) {
+        return String(e?.message)
+    }
+}
+
+async function checkUnderlyingWalletHealth(
+    wallet: IBlockChainWallet
+): Promise<string | undefined> {
+    try {
+        const resp = await wallet.getTransactionFee({ isPayment: true })
+        if (resp.lten(0)) {
+            return 'client returned invalid response'
+        }
+    } catch (e: any) {
+        return String(e?.message)
+    }
+}
+
+async function serviceHealthcheck(
+    config: AgentBotConfig,
+    notifier: AgentNotifier
+) {
+    const errors: [string, string][] = []
+    for (const c of config.fAssets.values()) {
+        const rpcHealth = await checkNativeRpcHealth()
+        const indexerHealth = await checkIndexerHealth(c.blockchainIndexerClient)
+        // @ts-ignore flareDataConnector has the right implementation
+        const dalHealth = await checkDalHealth(c.flareDataConnector)
+        const walletHealth = await checkUnderlyingWalletHealth(c.wallet)
+        if (rpcHealth != null) {
+            errors.push(['native rpc', rpcHealth])
+        } else if (indexerHealth != null) {
+            errors.push(['FDC verifier', indexerHealth])
+        } else if (dalHealth != null) {
+            errors.push(['DAL', dalHealth])
+        } else if (walletHealth != null) {
+            errors.push(['underlying rpc', walletHealth])
+        }
+    }
+    for (const [service, reason] of errors) {
+        notifier.serviceUnhealthy(service, reason)
+    }
+}
+
+let healthcheckMonitorTimer: NodeJS.Timeout | null = null;
+
+export function startHealthcheckMonitor(
+    config: AgentBotConfig,
+    notifier: AgentNotifier,
+    healthcheckInterval: number
+) {
+    void serviceHealthcheck(config, notifier);
+    healthcheckMonitorTimer = setInterval(
+        () => void serviceHealthcheck(config, notifier),
+        healthcheckInterval
+    );
+}
+
+export function stopHealthcheckMonitor() {
+    if (healthcheckMonitorTimer) {
+        clearInterval(healthcheckMonitorTimer);
+        logger.info("Healthcheck monitor timer was cleared.");
+        console.log("Healthcheck monitor timer was cleared.");
+    }
+}

packages/fasset-bots-core/src/index.ts

@@ -26,6 +26,7 @@ export * from "./config/config-file-loader";
 export { encryptText, decryptText, EncryptionMethod } from "@flarenetwork/simple-wallet";
 export { isJSON, promptForPassword } from "./utils/prompt";
 export { startActivityTimestampUpdater, stopActivityTimestampUpdater, lastActivityTimestampSeconds } from "./actors/activity-updater";
+export { startHealthcheckMonitor, stopHealthcheckMonitor } from "./actors/service-health-checker";
 export * from "./fasset/Conversions";
 export * from "./state/CollateralPrice";
 export * from "./underlying-chain/interfaces/IBlockChain";

packages/fasset-bots-core/src/underlying-chain/BlockchainIndexerHelper.ts

@@ -202,6 +202,11 @@ export class BlockchainIndexerHelper implements IBlockChain {
         return blockHeight;
     }
 
+    async getHealth(): Promise<boolean> {
+        const respdata = await this.verifier.get<ApiWrapper<boolean>>(`/api/health`, "getHealth");
+        return respdata.status == "OK";
+    }
+
     async getCurrentBlockHeightFromIndexer(): Promise<number> {
         const respdata = await this.verifier.get<ApiWrapper<number>>(`/api/indexer/block-height-tip`, "getCurrentBlockHeightFromIndexer");
         const status = respdata.status;

packages/fasset-bots-core/src/underlying-chain/FlareDataConnectorClientHelper.ts

@@ -3,7 +3,7 @@ import {
     ConfirmedBlockHeightExists, Payment, ReferencedPaymentNonexistence, decodeAttestationName
 } from "@flarenetwork/state-connector-protocol";
 import { IFdcHubInstance, IFdcRequestFeeConfigurationsInstance, IFdcVerificationInstance, IRelayInstance } from "../../typechain-truffle";
-import { FspStatusResult } from "../utils/data-access-layer-types";
+import { DalHealthResult, FspStatusResult } from "../utils/data-access-layer-types";
 import { findRequiredEvent } from "../utils/events/truffle";
 import { formatArgs } from "../utils/formatting";
 import { DEFAULT_RETRIES, ZERO_BYTES32, retry, sleep } from "../utils/helpers";
@@ -257,6 +257,13 @@ export class FlareDataConnectorClientHelper implements IFlareDataConnectorClient
         return proof;
     }
 
+    async getHealth(): Promise<boolean[]> {
+        const responses = await Promise.all(this.dataAccessLayerClients.map(
+            client => client.get<DalHealthResult>('/api/health', 'getHealth', 0))
+        );
+        return responses.map(r => r.healthy)
+    }
+
     /* istanbul ignore next */
     private async verifyProof(proofData: AttestationProof): Promise<boolean> {
         const normalizedProofData = web3DeepNormalize(proofData);

packages/fasset-bots-core/src/utils/data-access-layer-types.ts

@@ -21,3 +21,7 @@ export interface FtsoFeedResult {
     turnoutBIPS: number | string;
     decimals: number | string;
 }
+
+export interface DalHealthResult {
+    healthy: boolean;
+}

packages/fasset-bots-core/src/utils/notifier/AgentNotifier.ts

@@ -118,6 +118,7 @@ export enum AgentNotificationKey {
     // unexpected errors - typically external service or network errors
     GENERIC_INFO = "INFO",
     UNEXPECTED_ERROR = "UNEXPECTED ERROR",
+    SERVICE_UNHEALTHY = "SERVICE UNHEALTHY",
 }
 
 export const agentNotifierThrottlingTimes: NotifierThrottlingConfigs = {
@@ -142,6 +143,10 @@ export class AgentNotifier extends BaseNotifier<AgentNotificationKey> {
         await this.danger(AgentNotificationKey.UNEXPECTED_ERROR, fullMessage);
     }
 
+    async serviceUnhealthy(service: string, reason: string) {
+        await this.danger(AgentNotificationKey.SERVICE_UNHEALTHY, `Service ${service} is unhealthy: "${reason}".`)
+    }
+
     async agentCreationFailed(error: string) {
         await this.danger(AgentNotificationKey.AGENT_CREATED_ERROR, `Failed to create agent: ${error}.`);
     }