Add metrics for loading data from storage

Author: DZakh

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

@@ -241,25 +241,6 @@ module BenchmarkCounters = {
   }
 }
 
-module PartitionBlockFetched = {
-  type labels = {chainId: int, partitionId: string}
-
-  let labelSchema = S.schema(s => {
-    chainId: s.matches(S.string->S.coerce(S.int)),
-    partitionId: s.matches(S.string),
-  })
-
-  let counter = SafeGauge.makeOrThrow(
-    ~name="partition_block_fetched",
-    ~help="The latest fetched block number for each partition",
-    ~labelSchema,
-  )
-
-  let set = (~blockNumber, ~partitionId, ~chainId) => {
-    counter->SafeGauge.handleInt(~labels={chainId, partitionId}, ~value=blockNumber)
-  }
-}
-
 let chainIdLabelsSchema = S.object(s => {
   s.field("chainId", S.string->S.coerce(S.int))
 })
@@ -440,22 +421,13 @@ module SourceGetHeightDuration = {
 }
 
 module ReorgCount = {
-  let deprecatedCounter = PromClient.Counter.makeCounter({
-    "name": "reorgs_detected",
-    "help": "Total number of reorgs detected",
-    "labelNames": ["chainId"],
-  })
-
   let gauge = SafeGauge.makeOrThrow(
     ~name="envio_reorg_count",
     ~help="Total number of reorgs detected",
     ~labelSchema=chainIdLabelsSchema,
   )
 
   let increment = (~chain) => {
-    deprecatedCounter
-    ->PromClient.Counter.labels({"chainId": chain->ChainMap.Chain.toString})
-    ->PromClient.Counter.inc
     gauge->SafeGauge.increment(~labels=chain->ChainMap.Chain.toChainId)
   }
 }
@@ -642,3 +614,79 @@ module EffectCacheCount = {
     gauge->SafeGauge.handleInt(~labels=effectName, ~value=count)
   }
 }
+
+module StorageLoad = {
+  let operationLabelsSchema = S.object(s => s.field("operation", S.string))
+
+  let timeCounter = SafeCounter.makeOrThrow(
+    ~name="envio_storage_load_time",
+    ~help="Processing time taken to load data from storage. (milliseconds)",
+    ~labelSchema=operationLabelsSchema,
+  )
+
+  let totalTimeCounter = SafeCounter.makeOrThrow(
+    ~name="envio_storage_load_total_time",
+    ~help="Cumulative time spent loading data from storage during the indexing process. (milliseconds)",
+    ~labelSchema=operationLabelsSchema,
+  )
+
+  let counter = SafeCounter.makeOrThrow(
+    ~name="envio_storage_load_count",
+    ~help="Cumulative number of successful storage load operations during the indexing process.",
+    ~labelSchema=operationLabelsSchema,
+  )
+
+  let whereSizeCounter = SafeCounter.makeOrThrow(
+    ~name="envio_storage_load_where_size",
+    ~help="Cumulative number of filter conditions ('where' items) used in storage load operations during the indexing process.",
+    ~labelSchema=operationLabelsSchema,
+  )
+
+  let sizeCounter = SafeCounter.makeOrThrow(
+    ~name="envio_storage_load_size",
+    ~help="Cumulative number of records loaded from storage during the indexing process.",
+    ~labelSchema=operationLabelsSchema,
+  )
+
+  type operationRef = {
+    mutable pendingCount: int,
+    timerRef: Hrtime.timeRef,
+  }
+  let operations = Js.Dict.empty()
+
+  let startOperation = (~operation) => {
+    switch operations->Utils.Dict.dangerouslyGetNonOption(operation) {
+    | Some(operationRef) => operationRef.pendingCount = operationRef.pendingCount + 1
+    | None =>
+      operations->Js.Dict.set(
+        operation,
+        (
+          {
+            pendingCount: 1,
+            timerRef: Hrtime.makeTimer(),
+          }: operationRef
+        ),
+      )
+    }
+    Hrtime.makeTimer()
+  }
+
+  let endOperation = (timerRef, ~operation, ~whereSize, ~size) => {
+    let operationRef = operations->Js.Dict.unsafeGet(operation)
+    operationRef.pendingCount = operationRef.pendingCount - 1
+    if operationRef.pendingCount === 0 {
+      timeCounter->SafeCounter.handleInt(
+        ~labels={operation},
+        ~value=operationRef.timerRef->Hrtime.timeSince->Hrtime.toMillis->Hrtime.intFromMillis,
+      )
+      operations->Utils.Dict.deleteInPlace(operation)
+    }
+    totalTimeCounter->SafeCounter.handleInt(
+      ~labels={operation},
+      ~value=timerRef->Hrtime.timeSince->Hrtime.toMillis->Hrtime.intFromMillis,
+    )
+    counter->SafeCounter.increment(~labels={operation})
+    whereSizeCounter->SafeCounter.handleInt(~labels={operation}, ~value=whereSize)
+    sizeCounter->SafeCounter.handleInt(~labels={operation}, ~value=size)
+  }
+}

codegenerator/cli/templates/static/codegen/src/LoadLayer.res

@@ -13,6 +13,8 @@ let loadById = (
   let inMemTable = inMemoryStore->InMemoryStore.getInMemTable(~entityConfig)
 
   let load = async idsToLoad => {
+    let timerRef = Prometheus.StorageLoad.startOperation(~operation=key)
+
     // Since LoadManager.call prevents registerign entities already existing in the inMemoryStore,
     // we can be sure that we load only the new ones.
     let dbEntities = try {
@@ -41,6 +43,12 @@ let loadById = (
         ~entity=entitiesMap->Utils.Dict.dangerouslyGetNonOption(entityId),
       )
     })
+
+    timerRef->Prometheus.StorageLoad.endOperation(
+      ~operation=key,
+      ~whereSize=idsToLoad->Array.length,
+      ~size=dbEntities->Array.length,
+    )
   }
 
   loadManager->LoadManager.call(
@@ -158,6 +166,10 @@ let loadByField = (
   let inMemTable = inMemoryStore->InMemoryStore.getInMemTable(~entityConfig)
 
   let load = async (fieldValues: array<'fieldValue>) => {
+    let timerRef = Prometheus.StorageLoad.startOperation(~operation=key)
+
+    let size = ref(0)
+
     let indiciesToLoad = fieldValues->Js.Array2.map((fieldValue): TableIndices.Index.t => {
       Single({
         fieldName,
@@ -210,8 +222,16 @@ let loadByField = (
             ~entity=Some(entity),
           )
         })
+
+        size := size.contents + entities->Array.length
       })
       ->Promise.all
+
+    timerRef->Prometheus.StorageLoad.endOperation(
+      ~operation=key,
+      ~whereSize=fieldValues->Array.length,
+      ~size=size.contents,
+    )
   }
 
   loadManager->LoadManager.call(

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

@@ -383,16 +383,6 @@ let validatePartitionQueryResponse = (
   }
 
   if Env.Benchmark.shouldSaveData {
-    switch query.target {
-    | Merge(_) => ()
-    | Head
-    | EndBlock(_) =>
-      Prometheus.PartitionBlockFetched.set(
-        ~blockNumber=latestFetchedBlockNumber,
-        ~partitionId=query.partitionId,
-        ~chainId=chain->ChainMap.Chain.toChainId,
-      )
-    }
     Benchmark.addBlockRangeFetched(
       ~totalTimeElapsed=stats.totalTimeElapsed,
       ~parsingTimeElapsed=stats.parsingTimeElapsed->Belt.Option.getWithDefault(0),