sets up tests for new chainmetrics

codexcrawler/components/chainmetrics.nim

@@ -6,6 +6,7 @@ import pkg/questionable/results
 import ../state
 import ../services/metrics
 import ../services/marketplace
+import ../components/requeststore
 import ../component
 
 logScope:
@@ -14,10 +15,10 @@ logScope:
 type ChainMetrics* = ref object of Component
   state: State
   metrics: Metrics
+  store: RequestStore
   marketplace: MarketplaceService
 
 proc step(c: ChainMetrics): Future[?!void] {.async: (raises: []).} =
-  # replace slotFills entirely:
   # iterate all requests in requestStore:
     # get state of request on chain
     # if failed/canceled/error:
@@ -29,14 +30,6 @@ proc step(c: ChainMetrics): Future[?!void] {.async: (raises: []).} =
         # total num slots
         # total size of request
   # iter finished: update metrics!
-
-
-  # without slotFills =? (await c.marketplace.getRecentSlotFillEvents()), err:
-  #   trace "Unable to get recent slotFill events from chain", err = err.msg
-  #   return success() # We don't propagate this error.
-  #   # The call is allowed to fail and the app should continue as normal.
-
-  # c.metrics.setSlotFill(slotFills.len)
   return success()
 
 method start*(c: ChainMetrics): Future[?!void] {.async.} =
@@ -57,6 +50,7 @@ proc new*(
     T: type ChainMetrics,
     state: State,
     metrics: Metrics,
-    marketplace: MarketplaceService,
+    store: RequestStore,
+    marketplace: MarketplaceService
 ): ChainMetrics =
-  ChainMetrics(state: state, metrics: metrics, marketplace: marketplace)
+  ChainMetrics(state: state, metrics: metrics, store: store, marketplace: marketplace)

codexcrawler/services/marketplace.nim

@@ -18,10 +18,25 @@ type
     market: ?OnChainMarket
     clock: Clock
   OnNewRequest* = proc(id: Rid): Future[?!void] {.async: (raises: []), gcsafe.}
+  RequestInfo* = ref object
+    slots*: uint64
+    slotSize*: uint64
 
 proc notStarted() =
   raiseAssert("MarketplaceService was called before it was started.")
 
+proc fetchRequestInfo(market: OnChainMarket, rid: Rid): Future[?RequestInfo] {.async: (raises: []).} =
+  try:
+    let request = await market.getRequest(rid)
+    if r =? request:
+      return some(RequestInfo(
+        slots: r.ask.slots,
+        slotSize: r.ask.slotSize
+      ))
+  except CatchableError as exc:
+    trace "Failed to get request info", err = exc.msg
+  return none(RequestInfo)
+
 method subscribeToNewRequests*(m: MarketplaceService, onNewRequest: OnNewRequest): Future[?!void] {.async: (raises: []), base.} =
   proc resultWrapper(rid: Rid): Future[void] {.async.} =
     let response = await onNewRequest(rid)
@@ -57,6 +72,21 @@ method iteratePastNewRequestEvents*(m: MarketplaceService, onNewRequest: OnNewRe
   else:
     notStarted()
 
+method getRequestInfo*(m: MarketplaceService, rid: Rid): Future[?RequestInfo] {.async: (raises: []), base.} =
+  # If the request id exists and is running, fetch the request object and return the info object.
+  # otherwise, return none.
+  if market =? m.market:
+    try:
+      let state = await market.requestState(rid)
+      if s =? state:
+        if s == RequestState.Started:
+          return await market.fetchRequestInfo(rid)
+    except CatchableError as exc:
+      trace "Failed to get request state", err = exc.msg
+    return none(RequestInfo)
+  else:
+    notStarted()
+
 method start*(m: MarketplaceService): Future[?!void] {.async.} =
   let provider = JsonRpcProvider.new(m.state.config.ethProvider)
   without marketplaceAddress =? Address.init(m.state.config.marketplaceAddress):

codexcrawler/services/metrics.nim

@@ -6,7 +6,9 @@ declareGauge(todoNodesGauge, "DHT nodes to be visited")
 declareGauge(okNodesGauge, "DHT nodes successfully contacted")
 declareGauge(nokNodesGauge, "DHT nodes failed to contact")
 
-declareGauge(slotFillGauge, "Marketplace recent slots filled")
+declareGauge(requestsGauge, "Marketplace active storage requests")
+declareGauge(requestSlotsGauge, "Marketplace active storage request slots")
+declareGauge(totalStorageSizeGauge, "Marketplace total bytes stored in active storage requests")
 
 type
   OnUpdateMetric = proc(value: int64): void {.gcsafe, raises: [].}
@@ -15,7 +17,10 @@ type
     todoNodes: OnUpdateMetric
     okNodes: OnUpdateMetric
     nokNodes: OnUpdateMetric
-    onSlotFill: OnUpdateMetric
+
+    onRequests: OnUpdateMetric
+    onRequestSlots: OnUpdateMetric
+    onTotalSize: OnUpdateMetric
 
 proc startServer(metricsAddress: IpAddress, metricsPort: Port) =
   let metricsAddress = metricsAddress
@@ -37,8 +42,14 @@ method setOkNodes*(m: Metrics, value: int) {.base, gcsafe, raises: [].} =
 method setNokNodes*(m: Metrics, value: int) {.base, gcsafe, raises: [].} =
   m.nokNodes(value.int64)
 
-method setSlotFill*(m: Metrics, value: int) {.base, gcsafe, raises: [].} =
-  m.onSlotFill(value.int64)
+method setRequests*(m: Metrics, value: int) {.base, gcsafe, raises: [].} =
+  m.onRequests(value.int64)
+
+method setRequestSlots*(m: Metrics, value: int) {.base, gcsafe, raises: [].} =
+  m.onRequestSlots(value.int64)
+
+method setTotalSize*(m: Metrics, value: int) {.base, gcsafe, raises: [].} =
+  m.onTotalSize(value.int64)
 
 proc createMetrics*(metricsAddress: IpAddress, metricsPort: Port): Metrics =
   startServer(metricsAddress, metricsPort)
@@ -54,8 +65,14 @@ proc createMetrics*(metricsAddress: IpAddress, metricsPort: Port): Metrics =
   proc onNok(value: int64) =
     nokNodesGauge.set(value)
 
-  proc onSlotFill(value: int64) =
-    slotFillGauge.set(value)
+  proc onRequests(value: int64) =
+    requestsGauge.set(value)
+
+  proc onRequestSlots(value: int64) =
+    requestSlotsGauge.set(value)
+
+  proc onTotalSize(value: int64) =
+    totalStorageSizeGauge.set(value)
 
   return
-    Metrics(todoNodes: onTodo, okNodes: onOk, nokNodes: onNok, onSlotFill: onSlotFill)
+    Metrics(todoNodes: onTodo, okNodes: onOk, nokNodes: onNok, onRequests: onRequests, onRequestSlots: onRequestSlots, onTotalSize: onTotalSize)

tests/codexcrawler/components/testchainmetrics.nim

@@ -1,85 +1,110 @@
-# import pkg/chronos
-# import pkg/questionable
-# import pkg/questionable/results
-# import pkg/asynctest/chronos/unittest
-# import std/sequtils
+import pkg/chronos
+import pkg/questionable
+import pkg/questionable/results
+import pkg/asynctest/chronos/unittest
 
-# import ../../../codexcrawler/components/chainmetrics
-# import ../../../codexcrawler/services/marketplace/market
-# import ../../../codexcrawler/types
-# import ../../../codexcrawler/state
-# import ../mocks/mockstate
-# import ../mocks/mockmetrics
-# import ../mocks/mockmarketplace
-# import ../helpers
+import ../../../codexcrawler/components/chainmetrics
+import ../../../codexcrawler/components/requeststore
+import ../../../codexcrawler/services/marketplace
+import ../../../codexcrawler/types
+import ../mocks/mockstate
+import ../mocks/mockmetrics
+import ../mocks/mockrequeststore
+import ../mocks/mockmarketplace
+import ../helpers
 
-# suite "ChainMetrics":
-#   var
-#     state: MockState
-#     metrics: MockMetrics
-#     marketplace: MockMarketplaceService
-#     chain: ChainMetrics
+suite "ChainMetrics":
+  var
+    state: MockState
+    metrics: MockMetrics
+    store: MockRequestStore
+    marketplace: MockMarketplaceService
+    chain: ChainMetrics
 
-#   setup:
-#     state = createMockState()
-#     metrics = createMockMetrics()
-#     marketplace = createMockMarketplaceService()
+  setup:
+    state = createMockState()
+    metrics = createMockMetrics()
+    store = createMockRequestStore()
+    marketplace = createMockMarketplaceService()
 
-#     metrics.slotFill = -1
-#     chain = ChainMetrics.new(state, metrics, marketplace)
+    chain = ChainMetrics.new(state, metrics, store, marketplace)
 
-#     (await chain.start()).tryGet()
+    (await chain.start()).tryGet()
 
-#   teardown:
-#     (await chain.stop()).tryGet()
-#     state.checkAllUnsubscribed()
+  teardown:
+    (await chain.stop()).tryGet()
+    state.checkAllUnsubscribed()
 
-#   proc onStep() {.async.} =
-#     (await state.steppers[0]()).tryGet()
+  proc onStep() {.async.} =
+    (await state.steppers[0]()).tryGet()
 
-#   test "start should start stepper for 10 minutes":
-#     check:
-#       state.delays.len == 1
-#       state.delays[0] == 10.minutes
+  test "start should start stepper for 10 minutes":
+    check:
+      state.delays.len == 1
+      state.delays[0] == 10.minutes
 
-#   test "onStep is not activated when config.marketplaceEnable is false":
-#     # Recreate chainMetrics, reset mockstate:
-#     (await chain.stop()).tryGet()
-#     state.steppers = @[]
-#     # disable marketplace:
-#     state.config.marketplaceEnable = false
-#     (await chain.start()).tryGet()
+  # iterate all requests in requestStore:
+    # get state of request on chain
+    # if failed/canceled/error:
+      # if last-seen is old (1month?3months?)
+        # delete entry
+    # else (request is running):
+      # count:
+        # total running
+        # total num slots
+        # total size of request
+  # iter finished: update metrics!
 
-#     check:
-#       state.steppers.len == 0
+  test "onStep should remove non-running requests from request store":
+    let rid = genRid()
+    store.iterateEntries.add(RequestEntry(id: rid))
 
-#   test "step should not call setSlotFill when getRecentSlotFillEvents fails":
-#     let testValue = -123
-#     metrics.slotFill = testValue
+    marketplace.requestInfoReturns = none(RequestInfo)
 
-#     marketplace.recentSlotFillEventsReturn = seq[SlotFilled].failure("testfailure")
+    await onStep()
 
-#     await onStep()
+    check:
+      marketplace.requestInfoRid == rid
+      store.removeRid == rid
+    
+  test "onStep should count the number of active requests":
+    let rid1 = genRid()
+    let rid2 = genRid()
+    store.iterateEntries.add(RequestEntry(id: rid1))
+    store.iterateEntries.add(RequestEntry(id: rid2))
 
-#     check:
-#       metrics.slotFill == testValue
+    marketplace.requestInfoReturns = some(RequestInfo())
 
-#   test "step should setSlotFill to zero when getRecentSlotFillEvents returns empty seq":
-#     metrics.slotFill = -123
+    await onStep()
 
-#     marketplace.recentSlotFillEventsReturn = success(newSeq[SlotFilled]())
+    check:
+      metrics.requests == 2
 
-#     await onStep()
+  test "onStep should count the number of active slots":
+    let rid = genRid()
+    store.iterateEntries.add(RequestEntry(id: rid))
 
-#     check:
-#       metrics.slotFill == 0
+    let info = RequestInfo(
+      slots: 123
+    )
+    marketplace.requestInfoReturns = some(info)
 
-#   test "step should setSlotFill to the length of seq returned from getRecentSlotFillEvents":
-#     let fills = @[SlotFilled(), SlotFilled(), SlotFilled(), SlotFilled()]
+    await onStep()
 
-#     marketplace.recentSlotFillEventsReturn = success(fills)
+    check:
+      metrics.slots == info.slots.int
 
-#     await onStep()
+  test "onStep should count the total size of active slots":
+    let rid = genRid()
+    store.iterateEntries.add(RequestEntry(id: rid))
 
-#     check:
-#       metrics.slotFill == fills.len
+    let info = RequestInfo(
+      slots: 12,
+      slotSize: 23
+    )
+    marketplace.requestInfoReturns = some(info)
+
+    await onStep()
+
+    check:
+      metrics.totalSize == (info.slots * info.slotSize).int

tests/codexcrawler/helpers.nim

@@ -1,5 +1,4 @@
 import std/random
-import std/sequtils
 import std/typetraits
 import pkg/stint
 import pkg/stew/byteutils

tests/codexcrawler/mocks/mockmarketplace.nim

@@ -2,7 +2,7 @@ import pkg/ethers
 import pkg/questionable
 
 import ../../../codexcrawler/services/marketplace
-import ../../../codexcrawler/services/marketplace/market
+import ../../../codexcrawler/types
 
 logScope:
   topics = "marketplace"
@@ -10,6 +10,8 @@ logScope:
 type MockMarketplaceService* = ref object of MarketplaceService
   subNewRequestsCallback*: ?OnNewRequest
   iterRequestsCallback*: ?OnNewRequest
+  requestInfoReturns*: ?RequestInfo
+  requestInfoRid*: Rid
 
 method subscribeToNewRequests*(m: MockMarketplaceService, onNewRequest: OnNewRequest): Future[?!void] {.async: (raises: []).} =
   m.subNewRequestsCallback = some(onNewRequest)
@@ -19,6 +21,10 @@ method iteratePastNewRequestEvents*(m: MockMarketplaceService, onNewRequest: OnN
   m.iterRequestsCallback = some(onNewRequest)
   return success()
 
+method getRequestInfo*(m: MockMarketplaceService, rid: Rid): Future[?RequestInfo] {.async: (raises: []).} =
+  m.requestInfoRid = rid
+  return m.requestInfoReturns
+
 proc createMockMarketplaceService*(): MockMarketplaceService =
   MockMarketplaceService(
     subNewRequestsCallback: none(OnNewRequest),

tests/codexcrawler/mocks/mockmetrics.nim

@@ -4,7 +4,9 @@ type MockMetrics* = ref object of Metrics
   todo*: int
   ok*: int
   nok*: int
-  slotFill*: int
+  requests*: int
+  slots*: int
+  totalSize*: int
 
 method setTodoNodes*(m: MockMetrics, value: int) =
   m.todo = value
@@ -15,8 +17,14 @@ method setOkNodes*(m: MockMetrics, value: int) =
 method setNokNodes*(m: MockMetrics, value: int) =
   m.nok = value
 
-method setSlotFill*(m: MockMetrics, value: int) =
-  m.slotFill = value
+method setRequests*(m: MockMetrics, value: int) =
+  m.requests = value
+
+method setRequestSlots*(m: MockMetrics, value: int) =
+  m.slots = value
+
+method setTotalSize*(m: MockMetrics, value: int) =
+  m.totalSize = value
 
 proc createMockMetrics*(): MockMetrics =
   MockMetrics()