implements dht-metrics

codexcrawler/components/dhtmetrics.nim

@@ -6,10 +6,9 @@ import pkg/questionable/results
 import ./dht
 import ../list
 import ../state
-import ../component
-import ../types
-import ../utils/asyncdataevent
 import ../metrics
+import ../component
+import ../utils/asyncdataevent
 
 logScope:
   topics = "dhtmetrics"
@@ -18,34 +17,48 @@ type DhtMetrics* = ref object of Component
   state: State
   ok: List
   nok: List
+  sub: AsyncDataEventSubscription
+  metrics: Metrics
+
+proc handleCheckEvent(
+    d: DhtMetrics, event: DhtNodeCheckEventData
+): Future[?!void] {.async.} =
+  if event.isOk:
+    ?await d.ok.add(event.id)
+    ?await d.nok.remove(event.id)
+  else:
+    ?await d.ok.remove(event.id)
+    ?await d.nok.add(event.id)
+
+  d.metrics.setOkNodes(d.ok.len)
+  d.metrics.setNokNodes(d.nok.len)
+
+  return success()
 
 method start*(d: DhtMetrics): Future[?!void] {.async.} =
   info "Starting DhtMetrics..."
+  ?await d.ok.load()
+  ?await d.nok.load()
+
+  proc onCheck(event: DhtNodeCheckEventData): Future[?!void] {.async.} =
+    await d.handleCheckEvent(event)
+
+  d.sub = d.state.events.dhtNodeCheck.subscribe(onCheck)
   return success()
 
 method stop*(d: DhtMetrics): Future[?!void] {.async.} =
+  await d.state.events.dhtNodeCheck.unsubscribe(d.sub)
   return success()
 
 proc new*(
-    T: type DhtMetrics,
-    state: State,
-    okList: List,
-    nokList: List
+    T: type DhtMetrics, state: State, okList: List, nokList: List, metrics: Metrics
 ): DhtMetrics =
-  DhtMetrics(
-    state: state,
-    ok: okList,
-    nok: nokList
-  )
+  DhtMetrics(state: state, ok: okList, nok: nokList, metrics: metrics)
 
-proc createDhtMetrics*(state: State): ?!DhtMetrics =
+proc createDhtMetrics*(state: State, metrics: Metrics): ?!DhtMetrics =
   without okList =? createList(state.config.dataDir, "dhtok"), err:
     return failure(err)
   without nokList =? createList(state.config.dataDir, "dhtnok"), err:
     return failure(err)
 
-  success(DhtMetrics.new(
-    state,
-    okList,
-    nokList
-  ))
+  success(DhtMetrics.new(state, okList, nokList, metrics))

codexcrawler/list.nim

@@ -19,12 +19,11 @@ import ./utils/datastoreutils
 logScope:
   topics = "list"
 
-type
-  List* = ref object of RootObj
-    name: string
-    store: TypedDatastore
-    items: HashSet[Nid]
-    emptySignal: ?Future[void]
+type List* = ref object of RootObj
+  name: string
+  store: TypedDatastore
+  items: HashSet[Nid]
+  emptySignal: ?Future[void]
 
 proc encode(s: Nid): seq[byte] =
   s.toBytes()
@@ -86,12 +85,10 @@ method remove*(this: List, nid: Nid): Future[?!void] {.async, base.} =
   ?await this.store.delete(itemKey)
   return success()
 
-proc len*(this: List): int =
+method len*(this: List): int {.base, gcsafe, raises: [].} =
   this.items.len
 
-proc new*(
-    _: type List, name: string, store: TypedDatastore
-): List =
+proc new*(_: type List, name: string, store: TypedDatastore): List =
   List(name: name, store: store)
 
 proc createList*(dataDir: string, name: string): ?!List =

codexcrawler/metrics.nim

@@ -9,7 +9,7 @@ declareGauge(nokNodesGauge, "DHT nodes failed to contact")
 type
   OnUpdateMetric = proc(value: int64): void {.gcsafe, raises: [].}
 
-  Metrics* = ref object
+  Metrics* = ref object of RootObj
     todoNodes: OnUpdateMetric
     okNodes: OnUpdateMetric
     nokNodes: OnUpdateMetric
@@ -25,13 +25,13 @@ proc startServer(metricsAddress: IpAddress, metricsPort: Port) =
   except Exception as exc:
     raiseAssert exc.msg # TODO fix metrics
 
-method setTodoNodes*(m: Metrics, value: int) {.base.} =
+method setTodoNodes*(m: Metrics, value: int) {.base, gcsafe, raises: [].} =
   m.todoNodes(value.int64)
 
-method setOkNodes*(m: Metrics, value: int) {.base.} =
+method setOkNodes*(m: Metrics, value: int) {.base, gcsafe, raises: [].} =
   m.okNodes(value.int64)
 
-method setNokNodes*(m: Metrics, value: int) {.base.} =
+method setNokNodes*(m: Metrics, value: int) {.base, gcsafe, raises: [].} =
   m.nokNodes(value.int64)
 
 proc createMetrics*(metricsAddress: IpAddress, metricsPort: Port): Metrics =
@@ -47,9 +47,5 @@ proc createMetrics*(metricsAddress: IpAddress, metricsPort: Port): Metrics =
 
   proc onNok(value: int64) =
     nokNodesGauge.set(value)
-  
-  return Metrics(
-    todoNodes: onTodo,
-    okNodes: onOk,
-    nokNodes: onNok
-  )
+
+  return Metrics(todoNodes: onTodo, okNodes: onOk, nokNodes: onNok)

tests/codexcrawler/components/testdhtmetrics.nim

@@ -1,8 +1,6 @@
-import std/os
 import pkg/chronos
 import pkg/questionable/results
 import pkg/asynctest/chronos/unittest
-import pkg/datastore/typedds
 
 import ../../../codexcrawler/components/dhtmetrics
 import ../../../codexcrawler/utils/asyncdataevent
@@ -10,6 +8,7 @@ import ../../../codexcrawler/types
 import ../../../codexcrawler/state
 import ../mockstate
 import ../mocklist
+import ../mockmetrics
 import ../helpers
 
 suite "DhtMetrics":
@@ -18,6 +17,7 @@ suite "DhtMetrics":
     state: MockState
     okList: MockList
     nokList: MockList
+    metrics: MockMetrics
     dhtmetrics: DhtMetrics
 
   setup:
@@ -25,12 +25,9 @@ suite "DhtMetrics":
     state = createMockState()
     okList = createMockList()
     nokList = createMockList()
+    metrics = createMockMetrics()
 
-    dhtmetrics = DhtMetrics.new(
-      state,
-      okList,
-      nokList
-    )
+    dhtmetrics = DhtMetrics.new(state, okList, nokList, metrics)
 
     (await dhtmetrics.start()).tryGet()
 
@@ -39,17 +36,11 @@ suite "DhtMetrics":
     state.checkAllUnsubscribed()
 
   proc fireDhtNodeCheckEvent(isOk: bool) {.async.} =
-    let
-      event = DhtNodeCheckEventData(
-        id: nid,
-        isOk: isOk
-      )
+    let event = DhtNodeCheckEventData(id: nid, isOk: isOk)
 
     (await state.events.dhtNodeCheck.fire(event)).tryGet()
 
   test "dhtmetrics start should load both lists":
-    (await dhtmetrics.start()).tryGet()
-
     check:
       okList.loadCalled
       nokList.loadCalled
@@ -58,13 +49,13 @@ suite "DhtMetrics":
     await fireDhtNodeCheckEvent(true)
 
     check:
-      nid in okList.added 
+      nid in okList.added
 
   test "dhtNodeCheck event should add node to nokList if check has failed":
     await fireDhtNodeCheckEvent(false)
 
     check:
-      nid in nokList.added 
+      nid in nokList.added
 
   test "dhtNodeCheck event should remove node from nokList if check is successful":
     await fireDhtNodeCheckEvent(true)
@@ -77,3 +68,23 @@ suite "DhtMetrics":
 
     check:
       nid in okList.removed
+
+  test "dhtNodeCheck event should set okList length as dht-ok metric":
+    let length = 123
+
+    okList.length = length
+
+    await fireDhtNodeCheckEvent(true)
+
+    check:
+      metrics.ok == length
+
+  test "dhtNodeCheck event should set nokList length as dht-nok metric":
+    let length = 234
+
+    nokList.length = length
+
+    await fireDhtNodeCheckEvent(true)
+
+    check:
+      metrics.nok == length

tests/codexcrawler/mocklist.nim

@@ -4,16 +4,17 @@ import pkg/questionable/results
 import ../../codexcrawler/types
 import ../../codexcrawler/list
 
-type
-  MockList* = ref object of List
-    loadCalled*: bool
-    added*: seq[Nid]
-    addSuccess*: bool
-    removed*: seq[Nid]
-    removeSuccess*: bool
+type MockList* = ref object of List
+  loadCalled*: bool
+  added*: seq[Nid]
+  addSuccess*: bool
+  removed*: seq[Nid]
+  removeSuccess*: bool
+  length*: int
 
 method load*(this: MockList): Future[?!void] {.async.} =
   this.loadCalled = true
+  return success()
 
 method add*(this: MockList, nid: Nid): Future[?!void] {.async.} =
   this.added.add(nid)
@@ -27,11 +28,15 @@ method remove*(this: MockList, nid: Nid): Future[?!void] {.async.} =
     return success()
   return failure("test failure")
 
+method len*(this: MockList): int =
+  return this.length
+
 proc createMockList*(): MockList =
   MockList(
     loadCalled: false,
     added: newSeq[Nid](),
     addSuccess: true,
     removed: newSeq[Nid](),
-    removeSuccess: true
+    removeSuccess: true,
+    length: 0,
   )

tests/codexcrawler/mockmetrics.nim

@@ -0,0 +1,18 @@
+import ../../codexcrawler/metrics
+
+type MockMetrics* = ref object of Metrics
+  todo*: int
+  ok*: int
+  nok*: int
+
+method setTodoNodes*(m: MockMetrics, value: int) =
+  m.todo = value
+
+method setOkNodes*(m: MockMetrics, value: int) =
+  m.ok = value
+
+method setNokNodes*(m: MockMetrics, value: int) =
+  m.nok = value
+
+proc createMockMetrics*(): MockMetrics =
+  MockMetrics()