Indexing performance metrics (#545)

* Add prom metrics to measure quering performance

* Update source height metric using query response

* Sync chain list

Author: DZakh

codegenerator/cli/npm/envio/src/Prometheus.res

@@ -83,6 +83,7 @@ module MakeSafePromMetric = (
   let handleInt: (t<'a>, ~labels: 'a, ~value: int) => unit
   let handleFloat: (t<'a>, ~labels: 'a, ~value: float) => unit
   let increment: (t<'a>, ~labels: 'a) => unit
+  let incrementMany: (t<'a>, ~labels: 'a, ~value: int) => unit
 } => {
   type t<'a> = {metric: M.t, labelSchema: S.t<'a>}
 
@@ -121,6 +122,13 @@ module MakeSafePromMetric = (
       ->M.labels(labels->S.reverseConvertToJsonOrThrow(labelSchema))
       ->Obj.magic
     )["inc"]()
+
+  let incrementMany = ({metric, labelSchema}: t<'a>, ~labels: 'a, ~value) =>
+    (
+      metric
+      ->M.labels(labels->S.reverseConvertToJsonOrThrow(labelSchema))
+      ->Obj.magic
+    )["inc"](value)
 }
 
 module SafeCounter = MakeSafePromMetric({
@@ -356,6 +364,30 @@ module IndexingPartitions = {
   }
 }
 
+module IndexingIdleTime = {
+  let counter = SafeCounter.makeOrThrow(
+    ~name="envio_indexing_idle_time",
+    ~help="The number of milliseconds the indexer source syncing has been idle. A high value may indicate the source sync is a bottleneck.",
+    ~labelSchema=chainIdLabelsSchema,
+  )
+}
+
+module IndexingSourceWaitingTime = {
+  let counter = SafeCounter.makeOrThrow(
+    ~name="envio_indexing_source_waiting_time",
+    ~help="The number of milliseconds the indexer has been waiting for new blocks.",
+    ~labelSchema=chainIdLabelsSchema,
+  )
+}
+
+module IndexingQueryTime = {
+  let counter = SafeCounter.makeOrThrow(
+    ~name="envio_indexing_query_time",
+    ~help="The number of milliseconds spent performing queries to the chain data-source.",
+    ~labelSchema=chainIdLabelsSchema,
+  )
+}
+
 module IndexingBufferSize = {
   let gauge = SafeGauge.makeOrThrow(
     ~name="envio_indexing_buffer_size",

codegenerator/cli/npm/envio/src/sources/SourceManager.res

@@ -1,12 +1,16 @@
 open Belt
 
+type sourceManagerStatus = Idle | WaitingForNewBlock | Querieng
+
 // Ideally the ChainFetcher name suits this better
 // But currently the ChainFetcher module is immutable
 // and handles both processing and fetching.
 // So this module is to encapsulate the fetching logic only
 // with a mutable state for easier reasoning and testing.
 type t = {
   sources: Utils.Set.t<Source.t>,
+  mutable statusStart: Hrtime.timeRef,
+  mutable status: sourceManagerStatus,
   maxPartitionConcurrency: int,
   newBlockFallbackStallTimeout: int,
   stalledPollingInterval: int,
@@ -66,7 +70,23 @@ let make = (
     newBlockFallbackStallTimeout,
     stalledPollingInterval,
     getHeightRetryInterval,
+    statusStart: Hrtime.makeTimer(),
+    status: Idle,
+  }
+}
+
+let trackNewStatus = (sourceManager: t, ~newStatus) => {
+  let promCounter = switch newStatus {
+  | Idle => Prometheus.IndexingIdleTime.counter
+  | WaitingForNewBlock => Prometheus.IndexingSourceWaitingTime.counter
+  | Querieng => Prometheus.IndexingQueryTime.counter
   }
+  promCounter->Prometheus.SafeCounter.incrementMany(
+    ~labels=sourceManager.activeSource.chain->ChainMap.Chain.toChainId,
+    ~value=sourceManager.statusStart->Hrtime.timeSince->Hrtime.toMillis->Hrtime.intFromMillis,
+  )
+  sourceManager.statusStart = Hrtime.makeTimer()
+  sourceManager.status = newStatus
 }
 
 let fetchNext = async (
@@ -96,10 +116,12 @@ let fetchNext = async (
     | Some(waitingStateId) if waitingStateId >= stateId => ()
     | Some(_) // Case for the prev state before a rollback
     | None =>
+      sourceManager->trackNewStatus(~newStatus=WaitingForNewBlock)
       sourceManager.waitingForNewBlockStateId = Some(stateId)
       let currentBlockHeight = await waitForNewBlock(~currentBlockHeight)
       switch sourceManager.waitingForNewBlockStateId {
       | Some(waitingStateId) if waitingStateId === stateId => {
+          sourceManager->trackNewStatus(~newStatus=Idle)
           sourceManager.waitingForNewBlockStateId = None
           onNewBlock(~currentBlockHeight)
         }
@@ -115,6 +137,7 @@ let fetchNext = async (
         ~concurrency=sourceManager.fetchingPartitionsCount,
         ~chainId=sourceManager.activeSource.chain->ChainMap.Chain.toChainId,
       )
+      sourceManager->trackNewStatus(~newStatus=Querieng)
       let _ =
         await queries
         ->Array.map(q => {
@@ -125,6 +148,9 @@ let fetchNext = async (
               ~concurrency=sourceManager.fetchingPartitionsCount,
               ~chainId=sourceManager.activeSource.chain->ChainMap.Chain.toChainId,
             )
+            if sourceManager.fetchingPartitionsCount === 0 {
+              sourceManager->trackNewStatus(~newStatus=Idle)
+            }
           })
           promise
         })

codegenerator/cli/src/config_parsing/chain_helpers.rs

@@ -250,7 +250,7 @@ pub enum Network {
     )]
     MoonbaseAlpha = 1287,
 
-    #[subenum(NetworkWithExplorer, GraphNetwork)]
+    #[subenum(HypersyncNetwork, NetworkWithExplorer, GraphNetwork)]
     Moonbeam = 1284,
 
     #[subenum(GraphNetwork, NetworkWithExplorer)]
@@ -340,6 +340,9 @@ pub enum Network {
     #[subenum(HypersyncNetwork)]
     Superseed = 5330,
 
+    #[subenum(HypersyncNetwork)]
+    Swell = 1923,
+
     #[subenum(NetworkWithExplorer)]
     Taiko = 167000,
 
@@ -522,7 +525,8 @@ impl Network {
             | Network::MegaethTestnet
             | Network::Curtis
             | Network::Worldchain
-            | Network::Sonic => DEFAULT_CONFIRMED_BLOCK_THRESHOLD,
+            | Network::Sonic
+            | Network::Swell => DEFAULT_CONFIRMED_BLOCK_THRESHOLD,
         }
     }
 }
@@ -582,7 +586,7 @@ impl HypersyncNetwork {
             | ArbitrumSepolia | Fraxtal | Soneium | BaseSepolia | MevCommit | Merlin | Mode
             | MoonbaseAlpha | XdcTestnet | Morph | Harmony | Saakuru | Cyber | Superseed
             | MegaethTestnet | Sonic | Worldchain | Sophon | Fantom | Sepolia | Rsk | Chiliz
-            | Lisk | Hyperliquid => Stone,
+            | Lisk | Hyperliquid | Swell | Moonbeam => Stone,
         }
     }
 

codegenerator/cli/templates/static/codegen/src/globalState/GlobalState.res

@@ -343,6 +343,17 @@ let validatePartitionQueryResponse = (
   } = response
   let {rangeLastBlock} = reorgGuard
 
+  if currentBlockHeight > chainFetcher.currentBlockHeight {
+    Prometheus.SourceHeight.set(
+      ~blockNumber=currentBlockHeight,
+      ~chainId=chainFetcher.chainConfig.chain->ChainMap.Chain.toChainId,
+      // The currentBlockHeight from response won't necessarily
+      // belong to the currently active source.
+      // But for simplicity, assume it does.
+      ~sourceName=(chainFetcher.sourceManager->SourceManager.getActiveSource).name,
+    )
+  }
+
   if Env.Benchmark.shouldSaveData {
     switch query.target {
     | Merge(_) => ()