Add messages sent metrics to kathy (#749)

* re-enable eth for kathy

* Better gitignore

* add prometheus dns config

* add prometheus metrics to kathy

* fix newline

* cleanup gitignore

* fix job name

* cleanup gitignore

* cleanup gitignore

* make metrics submission a lib

.gitignore

@@ -14,10 +14,8 @@ rust/tmp.env
 tmp.env
 .DS_STORE
 
-typescript/*/.env
 typescript/*/node_modules
-typescript/**/tsconfig.tsbuildinfo
-**/**/tsconfig.tsbuildinfo
+solidity/*/artifacts
 
 .yarn/install-state.gz
 .yarn/cache

typescript/infra/config/environments/mainnet/helloworld.ts

@@ -11,8 +11,10 @@ export const helloWorld: HelloWorldConfig<MainnetChains> = {
       tag: 'sha-0f9c0f9',
     },
     cronSchedule: '0 15 * * *', // Every day at 3:00 PM UTC
-    chainsToSkip: ['ethereum'],
+    chainsToSkip: [],
     runEnv: environment,
     namespace: environment,
+    prometheusPushGateway:
+      'http://prometheus-pushgateway.monitoring.svc.cluster.local:9091',
   },
 };

typescript/infra/config/environments/testnet2/helloworld.ts

@@ -14,5 +14,7 @@ export const helloWorld: HelloWorldConfig<TestnetChains> = {
     chainsToSkip: [],
     runEnv: environment,
     namespace: environment,
+    prometheusPushGateway:
+      'http://prometheus-pushgateway.monitoring.svc.cluster.local:9091',
   },
 };

typescript/infra/helm/helloworld-kathy/templates/cron-job.yaml

@@ -28,4 +28,5 @@ spec:
             env:
             - name: CHAINS_TO_SKIP
               value: {{ join "," .Values.chainsToSkip }}
-
+            - name: PROMETHEUS_PUSH_GATEWAY
+              value: {{ .Values.infra.prometheusPushGateway }}

typescript/infra/scripts/funding/fund-relayers-from-deployer.ts

@@ -1,16 +1,13 @@
 import { Console } from 'console';
 import { ethers } from 'ethers';
-import { Gauge, Pushgateway, Registry } from 'prom-client';
+import { Gauge, Registry } from 'prom-client';
 
-import {
-  ChainConnection,
-  ChainName,
-  CompleteChainMap,
-} from '@abacus-network/sdk';
+import { ChainConnection, CompleteChainMap } from '@abacus-network/sdk';
 
 import { AgentKey, ReadOnlyAgentKey } from '../../src/agents/agent';
 import { getRelayerKeys } from '../../src/agents/key-utils';
 import { KEY_ROLE_ENUM } from '../../src/agents/roles';
+import { submitMetrics } from '../../src/utils/metrics';
 import { readJSONAtPath } from '../../src/utils/utils';
 import {
   assertEnvironment,
@@ -43,12 +40,6 @@ const walletBalanceGauge = new Gauge({
 });
 metricsRegister.registerMetric(walletBalanceGauge);
 
-interface FunderBalance {
-  chain: ChainName;
-  address?: string;
-  balance: number;
-}
-
 // Min delta is 1/10 of the desired balance
 const MIN_DELTA_NUMERATOR = ethers.BigNumber.from(1);
 const MIN_DELTA_DENOMINATOR = ethers.BigNumber.from(10);
@@ -148,8 +139,6 @@ async function main() {
     : getRelayerKeys(agentConfig);
 
   const chains = relayerKeys.map((key) => key.chainName!);
-  const balances: FunderBalance[] = [];
-
   let failureOccurred = false;
 
   for (const chain of chains) {
@@ -183,16 +172,23 @@ async function main() {
         failureOccurred = true;
       }
     }
-    balances.push({
-      chain,
-      address: funderAddress,
-      balance: parseFloat(
-        ethers.utils.formatEther(await chainConnection.signer!.getBalance()),
-      ),
-    });
+    walletBalanceGauge
+      .labels({
+        chain,
+        wallet_address: funderAddress ?? 'unknown',
+        wallet_name: 'relayer-funder',
+        token_symbol: 'Native',
+        token_name: 'Native',
+        ...constMetricLabels,
+      })
+      .set(
+        parseFloat(
+          ethers.utils.formatEther(await chainConnection.signer!.getBalance()),
+        ),
+      );
   }
 
-  await submitFunderBalanceMetrics(balances);
+  await submitMetrics(metricsRegister);
 
   if (failureOccurred) {
     error('At least one failure occurred when funding relayers');
@@ -217,53 +213,13 @@ function getRelayerKeysFromSerializedAddressFile(path: string): AgentKey[] {
     .filter((key: AgentKey) => key.role === KEY_ROLE_ENUM.Relayer);
 }
 
-function getPushGateway(): Pushgateway | null {
-  const gatewayAddr = process.env['PROMETHEUS_PUSH_GATEWAY'];
-  if (gatewayAddr) {
-    return new Pushgateway(gatewayAddr, [], metricsRegister);
-  } else {
-    warn(
-      'Prometheus push gateway address was not defined; not publishing metrics.',
-    );
-    return null;
-  }
-}
-
-async function submitFunderBalanceMetrics(balances: FunderBalance[]) {
-  const gateway = getPushGateway();
-  if (!gateway) return;
-
-  for (const { chain, address, balance } of balances) {
-    walletBalanceGauge
-      .labels({
-        chain,
-        wallet_address: address ?? 'unknown',
-        wallet_name: 'relayer-funder',
-        token_symbol: 'Native',
-        token_name: 'Native',
-        ...constMetricLabels,
-      })
-      .set(balance);
-  }
-
-  const { resp, body } = await gateway.push({ jobName: 'relayer_funder' });
-  const statusCode =
-    typeof resp == 'object' && resp != null && 'statusCode' in resp
-      ? (resp as any).statusCode
-      : 'unknown';
-  log(`Prometheus metrics pushed to PushGateway`, {
-    statusCode,
-    body,
-  });
-}
-
 function log(message: string, data?: any) {
   logWithFunction(console.log, message, data);
 }
 
-function warn(message: string, data?: any) {
-  logWithFunction(console.warn, message, data);
-}
+// function warn(message: string, data?: any) {
+//   logWithFunction(console.warn, message, data);
+// }
 
 function error(message: string, data?: any) {
   logWithFunction(console.error, message, data);

typescript/infra/scripts/helloworld/kathy.ts

@@ -1,13 +1,37 @@
+import { Counter, Registry } from 'prom-client';
+
 import { HelloWorldApp } from '@abacus-network/helloworld';
 import { ChainName, Chains } from '@abacus-network/sdk';
 
+import { submitMetrics } from '../../src/utils/metrics';
 import { sleep } from '../../src/utils/utils';
 import { getCoreEnvironmentConfig, getEnvironment } from '../utils';
 
 import { getApp } from './utils';
 
+const constMetricLabels = {
+  // this needs to get set in main because of async reasons
+  abacus_deployment: '',
+  abacus_context: 'abacus',
+};
+
+const metricsRegister = new Registry();
+const messagesCount = new Counter({
+  name: 'abacus_kathy_messages',
+  help: 'Test messages which have been sent with',
+  registers: [metricsRegister],
+  labelNames: [
+    'origin',
+    'remote',
+    'status',
+    ...(Object.keys(constMetricLabels) as (keyof typeof constMetricLabels)[]),
+  ],
+});
+metricsRegister.registerMetric(messagesCount);
+
 async function main() {
   const environment = await getEnvironment();
+  constMetricLabels.abacus_deployment = environment;
   const coreConfig = getCoreEnvironmentConfig(environment);
   const app = await getApp(coreConfig);
   const chains = app.chains() as Chains[];
@@ -26,21 +50,30 @@ async function main() {
 
   const sources = chains.filter((chain) => !skip || !skip.includes(chain));
   for (const source of sources) {
-    for (const destination of sources.slice().filter((d) => d !== source)) {
+    for (const destination of sources.filter((d) => d !== source)) {
+      const labels = {
+        origin: source,
+        remote: destination,
+        ...constMetricLabels,
+      };
       try {
         await sendMessage(app, source, destination);
+        messagesCount.labels({ ...labels, status: 'success' }).inc();
       } catch (err) {
         console.error(
           `Error sending message from ${source} to ${destination}, continuing...`,
           `${err}`.replaceAll('\n', ' ## '),
         );
         failureOccurred = true;
+        messagesCount.labels({ ...labels, status: 'failure' }).inc();
       }
       // Sleep 500ms to avoid race conditions where nonces are reused
       await sleep(500);
     }
   }
 
+  await submitMetrics(metricsRegister);
+
   if (failureOccurred) {
     console.error('Failure occurred at least once');
     process.exit(1);

typescript/infra/src/config/helloworld.ts

@@ -8,6 +8,7 @@ export interface HelloWorldKathyConfig<Chain extends ChainName> {
   runEnv: string;
   namespace: string;
   chainsToSkip: Chain[];
+  prometheusPushGateway: string;
 }
 
 export interface HelloWorldConfig<Chain extends ChainName> {

typescript/infra/src/helloworld/kathy.ts

@@ -40,6 +40,9 @@ function getHelloworldKathyHelmValues<Chain extends ChainName>(
       repository: kathyConfig.docker.repo,
       tag: kathyConfig.docker.tag,
     },
+    infra: {
+      prometheusPushGateway: kathyConfig.prometheusPushGateway,
+    },
   };
 
   return helmifyValues(values);

typescript/infra/src/utils/metrics.ts

@@ -0,0 +1,27 @@
+import { Pushgateway, Registry } from 'prom-client';
+
+function getPushGateway(register: Registry): Pushgateway | null {
+  const gatewayAddr = process.env['PROMETHEUS_PUSH_GATEWAY'];
+  if (gatewayAddr) {
+    return new Pushgateway(gatewayAddr, [], register);
+  } else {
+    console.warn(
+      'Prometheus push gateway address was not defined; not publishing metrics.',
+    );
+    return null;
+  }
+}
+
+export async function submitMetrics(register: Registry) {
+  const gateway = getPushGateway(register);
+  if (!gateway) return;
+
+  const { resp } = await gateway.push({ jobName: 'kathy' });
+  const statusCode =
+    typeof resp == 'object' && resp != null && 'statusCode' in resp
+      ? (resp as any).statusCode
+      : 'unknown';
+  console.log(
+    `Prometheus metrics pushed to PushGateway with status ${statusCode}`,
+  );
+}