add metrics

ethereum/package.json

@@ -7,6 +7,8 @@
     "@poanet/solidity-flattener": "^3.0.8",
     "@typechain/ethers-v5": "^10.1.1",
     "@types/chai": "^4.3.3",
+    "@types/koa": "^2.13.5",
+    "@types/koa-router": "^7.4.4",
     "@types/mocha": "^9.1.1",
     "chai": "^4.3.6",
     "dotenv": "^16.0.3",
@@ -33,6 +35,9 @@
     "@improbable-eng/grpc-web-node-http-transport": "^0.15.0",
     "elliptic": "^6.5.4",
     "jsonfile": "^6.1.0",
+    "koa": "^2.14.1",
+    "koa-router": "^12.0.0",
+    "prom-client": "^14.2.0",
     "typescript": "^4.8.3"
   }
 }

ethereum/ts-scripts/mockIntegration/loadGeneration.ts

@@ -1,9 +1,26 @@
 import { ChainInfo, init, loadChains } from "../helpers/env"
 import { sendMessage, sleep } from "./messageUtils"
+import { Counter, register } from "prom-client"
+import Koa from "koa"
+import Router from "koa-router"
+
+const promPort = 3000
 
 init()
 const chains = loadChains()
 
+export const undeliveredMessages = new Counter({
+  name: "undelivered_messages",
+  help: "Counter for number of messages that were not delivered",
+  labelNames: ["sourceChain", "targetChain"],
+})
+
+export const deliveredMessages = new Counter({
+  name: "delivered_messages",
+  help: "Counter for number of messages that were successfully delivered",
+  labelNames: ["sourceChain", "targetChain"],
+})
+
 async function run() {
   const chainIntervalIdx = process.argv.findIndex((arg) => arg === "--chainInterval")
   const salvoIntervalIdx = process.argv.findIndex((arg) => arg === "--salvoInterval")
@@ -22,25 +39,22 @@ async function run() {
   }
 }
 
-async function perChain(chainInterval: number, salvoInterval: number) {
+async function perChain(chainIntervalMS: number, salvoIntervalMS: number) {
   console.log(`Sending test messages to and from each chain...`)
   for (let salvo = 0; true; salvo++) {
     console.log("")
     console.log(`Sending salvo ${salvo}`)
+
     for (let i = 0; i < chains.length; ++i) {
       const j = i === 0 ? chains.length - 1 : 0
-      try {
-        await sendMessage(chains[i], chains[j], false, true)
-      } catch (e) {
-        console.error(e)
-      }
-      await sleep(chainInterval)
+      await sendMessageAndReportMetric(chains[i], chains[j], chainIntervalMS)
     }
-    await sleep(salvoInterval)
+
+    await sleep(salvoIntervalMS)
   }
 }
 
-async function matrix(chainInterval: number, salvoInterval: number) {
+async function matrix(chainIntervalMS: number, salvoIntervalMS: number) {
   console.log(`Sending test messages to and from every combination of chains...`)
   for (let salvo = 0; true; salvo++) {
     console.log("")
@@ -48,17 +62,49 @@ async function matrix(chainInterval: number, salvoInterval: number) {
 
     for (let i = 0; i < chains.length; ++i) {
       for (let j = 0; i < chains.length; ++i) {
-        try {
-          await sendMessage(chains[i], chains[j], false, true)
-        } catch (e) {
-          console.error(e)
-        }
-        await sleep(chainInterval)
+        await sendMessageAndReportMetric(chains[i], chains[j], chainIntervalMS)
       }
     }
-    await sleep(salvoInterval)
+
+    await sleep(salvoIntervalMS)
   }
 }
 
+async function sendMessageAndReportMetric(
+  sourceChain: ChainInfo,
+  targetChain: ChainInfo,
+  chainInterval: number
+) {
+  try {
+    const notFound = await sendMessage(sourceChain, targetChain, false, true)
+    const counter = notFound ? undeliveredMessages : deliveredMessages
+    counter
+      .labels({
+        sourceChain: sourceChain.chainId,
+        targetChain: targetChain.chainId,
+      })
+      .inc()
+  } catch (e) {
+    console.error(e)
+  }
+  await sleep(chainInterval)
+}
+
+async function launchMetricsServer() {
+  const app = new Koa()
+  const router = new Router()
+
+  router.get("/metrics", async (ctx, next) => {
+    let metrics = await register.metrics()
+    ctx.body = metrics
+  })
+
+  app.use(router.allowedMethods())
+  app.use(router.routes())
+  app.listen(promPort, () =>
+    console.log(`Prometheus metrics running on port ${promPort}`)
+  )
+}
+
 console.log("Start!")
 run().then(() => console.log("Done!"))

ethereum/ts-scripts/mockIntegration/messageUtils.ts

@@ -15,7 +15,7 @@ export async function sendMessage(
   targetChain: ChainInfo,
   fetchSignedVaa: boolean = false,
   queryMessageOnTargetFlag: boolean = true
-) {
+): Promise<boolean | undefined> {
   console.log(
     `Sending message from chain ${sourceChain.chainId} to ${targetChain.chainId}...`
   )
@@ -68,12 +68,15 @@ export async function sendMessage(
   }
 
   if (queryMessageOnTargetFlag) {
-    await queryMessageOnTarget(sentMessage, targetChain)
+    return await queryMessageOnTarget(sentMessage, targetChain)
   }
   console.log("")
 }
 
-async function queryMessageOnTarget(sentMessage: string, targetChain: ChainInfo) {
+async function queryMessageOnTarget(
+  sentMessage: string,
+  targetChain: ChainInfo
+): Promise<boolean> {
   let messageHistory: string[][] = []
   const targetIntegration = getMockIntegration(targetChain)
 
@@ -91,8 +94,14 @@ async function queryMessageOnTarget(sentMessage: string, targetChain: ChainInfo)
   }
   console.log("")
 
+  if (notFound) {
+    console.log(`ERROR: Did not receive message!`)
+    return false
+  }
+
   console.log(`Received message: ${messageHistory[messageHistory.length - 1][0]}`)
   console.log(`Received messageHistory: ${messageHistory.join(", ")}`)
+  return true
 }
 
 export async function encodeEmitterAddress(

relayer_engine/src/plugin/src/plugin.ts

@@ -4,7 +4,6 @@ import {
   assertInt,
   CommonPluginEnv,
   ContractFilter,
-  dbg,
   getScopedLogger,
   ParsedVaaWithBytes,
   parseVaaWithBytes,
@@ -20,7 +19,6 @@ import { Logger } from "winston"
 import { PluginError } from "./utils"
 import { SignedVaa } from "@certusone/wormhole-sdk"
 import {
-  IWormhole,
   IWormhole__factory,
   RelayProvider__factory,
   LogMessagePublishedEvent,
@@ -36,7 +34,6 @@ import * as ethers from "ethers"
 import { Implementation__factory } from "@certusone/wormhole-sdk/lib/cjs/ethers-contracts"
 import * as grpcWebNodeHttpTransport from "@improbable-eng/grpc-web-node-http-transport"
 import { retryAsyncUntilDefined } from "ts-retry/lib/cjs/retry"
-import { hexToNativeStringAlgorand } from "@certusone/wormhole-sdk/lib/cjs/algorand"
 
 let PLUGIN_NAME: string = "GenericRelayerPlugin"