setup telemetry for block parse state and block proposal

package.json

@@ -14,6 +14,8 @@
     "@nestjs/common": "^8.4.7",
     "@nestjs/core": "^8.4.7",
     "@nestjs/platform-express": "^8.4.7",
+    "@sentry/node": "^7.109.0",
+    "@sentry/profiling-node": "^7.109.0",
     "@summa-tx/bitcoin-spv-js": "^4.0.2",
     "@transmute/did-key-bls12381": "^0.3.0-unstable.10",
     "@types/node": "^20.12.2",

src/index.ts

@@ -1,8 +1,10 @@
 import { ApiModule } from "./modules/api/index"
 import { CoreService } from "./services"
 import { NewCoreService } from "./services/new";
+import telemetry from "./telemetry";
 
 async function startup(): Promise<void> {
+  telemetry.start()
 
   const coreNew = new NewCoreService();
   const core = new CoreService({
@@ -11,6 +13,10 @@ async function startup(): Promise<void> {
   await core.start()
 
   await coreNew.init(core)
+
+  // consensusKey initialized in coreNew.init()
+  telemetry.setUserId(coreNew.consensusKey.id, process.env.HIVE_ACCOUNT)
+
   await coreNew.start()
 
   const api = new ApiModule(1337, core)
@@ -21,6 +27,7 @@ async function startup(): Promise<void> {
     await core.stop()
     await coreNew.stop()
     await api.stop()
+    await telemetry.stop()
     process.exit(code)
   };
 

src/services/chainBridge.ts

@@ -21,6 +21,7 @@ import { PayloadTooLargeException } from '@nestjs/common'
 import { loggers } from 'winston'
 import { YogaServer } from 'graphql-yoga'
 import { WithdrawFinalization } from '../types/coreTransactions'
+import telemetry from '../telemetry'
 
 
 export class ChainBridge {
@@ -1112,6 +1113,11 @@ export class ChainBridge {
           id: 'last_hb_processed'
         })
         if(stateHeader) {
+          telemetry.captureEvent('hive_sync_status', {
+            blockLag: this.self.newService.chainBridge.blockLag,
+            streamRate: Math.round(diff / 15),
+            parseLag: this.self.newService.chainBridge.streamParser.stream.calcHeight - stateHeader.val,
+          })
           this.self.logger.info(`blockLag blockLag=${this.self.newService.chainBridge.blockLag} streamRate=${Math.round(diff / 15)} parseLag=${this.self.newService.chainBridge.streamParser.stream.calcHeight - stateHeader.val}`)
         } else {
           this.self.logger.info(`blockLag`, {

src/services/new/types.ts

@@ -333,7 +333,7 @@ export type HiveTransactionDbRecord = {
     operations: AnyOperation[]
 }
 
-export type AnyOperation = IndexedKeysToTuple<ListOfOperations> & ListOfOperations
+export type AnyOperation = any //IndexedKeysToTuple<ListOfOperations> & ListOfOperations
 
 type IndexedKeysToTuple<T extends {0: any; 1: any}> = [T[0], T[1]]
 

src/services/new/utils/streamUtils.ts

@@ -8,7 +8,7 @@ import { HiveTransactionDbRecord } from '../types';
 export interface EventRecord {
     id: "hive_block"
     key: string | number
-    transactions: HiveTransactionDbRecord[]
+    transactions: any[]//HiveTransactionDbRecord[]
     block_id: string
     timestamp: Date
 }

src/services/new/witness/index.ts

@@ -18,6 +18,8 @@ import { computeKeyId, sortTransactions } from '../utils';
 import { MultisigSystem } from './multisig';
 import { BalanceKeeper } from './balanceKeeper';
 
+import telemetry from '../../../telemetry';
+
 const Constants = {
   block_version: 1
 }
@@ -502,6 +504,12 @@ export class WitnessServiceV2 {
     }
 
     async proposeBlock(block_height: number) {
+      telemetry.captureEvent(`block consensus ${block_height}`, {
+        block_height,
+        lastest_block: this.self.chainBridge.streamParser.stream.lastBlock,
+        proposal: true,
+        proposer: process.env.HIVE_ACCOUNT,
+      })
 
       const lastHeader = await this.blockHeaders.findOne({
 
@@ -761,6 +769,13 @@ export class WitnessServiceV2 {
           return e.bn === slotHeight && e.account === fromWitness.account
       })
 
+      telemetry.captureEvent(`block consensus ${slotHeight}`, {
+        block_height: slotHeight,
+        latest_block: block_height,
+        proposal: false,
+        proposer: fromWitness.account,
+      })
+
       if(!witnessSlot) {
         console.log('Witness.cadBlock validate #1 - witness not in current slot')
         return;

src/telemetry.ts

@@ -0,0 +1,67 @@
+import * as Sentry from '@sentry/node'
+import { nodeProfilingIntegration } from '@sentry/profiling-node'
+
+type Primitive = number | string | boolean | bigint | symbol | null | undefined
+
+const errorHandler = (err) => {
+  Sentry.captureException(err)
+}
+
+const FLUSH_INTERVAL = 60 * 1000
+
+const flusher = () => Sentry.flush()
+
+let flushInterval
+
+export default {
+  start() {
+    Sentry.init({
+      dsn: 'https://8997351a7510d343abfe7227b44b20d4@o4507035962179584.ingest.us.sentry.io/4507039719358464',
+      integrations: [nodeProfilingIntegration()],
+      // Performance Monitoring
+      tracesSampleRate: 1.0, //  Capture 100% of the transactions
+      // Set sampling rate for profiling - this is relative to tracesSampleRate
+      profilesSampleRate: 1.0,
+    })
+
+    process.on('unhandledRejection', errorHandler)
+    process.on('uncaughtException', errorHandler)
+    flushInterval = setInterval(flusher, FLUSH_INTERVAL)
+  },
+  stop() {
+    process.off('unhandledRejection', errorHandler)
+    process.off('uncaughtException', errorHandler)
+    clearInterval(flushInterval)
+    return Sentry.close()
+  },
+  setUserId(id: string, username: string) {
+    Sentry.getCurrentScope().setUser({ id, username })
+  },
+  captureMessage(msg: string, ignoreContext: boolean = true) {
+    if (ignoreContext) {
+      Sentry.runWithAsyncContext(() => {
+        Sentry.captureMessage(msg)
+      })
+    } else {
+      Sentry.captureMessage(msg)
+    }
+  },
+  captureError(err: Error, ignoreContext: boolean = false) {
+    if (ignoreContext) {
+      Sentry.runWithAsyncContext(() => {
+        Sentry.captureException(err)
+      })
+    } else {
+      Sentry.captureException(err)
+    }
+  },
+  captureEvent(eventId: string, info: Record<string, Primitive>, ignoreContext: boolean = true) {
+    if (ignoreContext) {
+      Sentry.runWithAsyncContext(() => {
+        Sentry.captureMessage(eventId, { tags: info })
+      })
+    } else {
+      Sentry.captureMessage(eventId, { tags: info })
+    }
+  },
+}