chore(networkmonitor): use nim-presto + add timeout (#1389)

This commit is contained in:
Alvaro Revuelta 2022-11-16 16:38:31 +01:00 committed by GitHub
parent f1ab1475db
commit 0bbcc260c9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 54 additions and 38 deletions

View File

@ -4,17 +4,18 @@ else:
{.push raises: [].} {.push raises: [].}
import import
std/[tables,strutils,times,sequtils,httpclient], std/[tables,strutils,times,sequtils],
chronicles, chronicles,
chronicles/topics_registry, chronicles/topics_registry,
chronos, chronos,
chronos/timer as ctime,
confutils, confutils,
eth/keys, eth/keys,
eth/p2p/discoveryv5/enr, eth/p2p/discoveryv5/enr,
libp2p/crypto/crypto, libp2p/crypto/crypto,
metrics, metrics,
metrics/chronos_httpserver, metrics/chronos_httpserver,
presto/[route, server], presto/[route, server, client],
stew/shims/net stew/shims/net
import import
@ -42,7 +43,7 @@ proc setDiscoveredPeersCapabilities(
proc setConnectedPeersMetrics(discoveredNodes: seq[Node], proc setConnectedPeersMetrics(discoveredNodes: seq[Node],
node: WakuNode, node: WakuNode,
timeout: chronos.Duration, timeout: chronos.Duration,
client: HttpClient, restClient: RestClientRef,
allPeers: CustomPeersTableRef) {.async.} = allPeers: CustomPeersTableRef) {.async.} =
let currentTime = $getTime() let currentTime = $getTime()
@ -81,13 +82,19 @@ proc setConnectedPeersMetrics(discoveredNodes: seq[Node],
allPeers[peerId].ip = ip allPeers[peerId].ip = ip
# get more info the peers from its ip address # get more info the peers from its ip address
let location = await ipToLocation(ip, client) var location: NodeLocation
if not location.isOk(): try:
# IP-API endpoints are now limited to 45 HTTP requests per minute
# TODO: As network grows, find a better way to now block the whole app
await sleepAsync(1400)
let response = await restClient.ipToLocation(ip)
location = response.data
except:
warn "could not get location", ip=ip warn "could not get location", ip=ip
continue continue
allPeers[peerId].country = location.get().country allPeers[peerId].country = location.country
allPeers[peerId].city = location.get().city allPeers[peerId].city = location.city
let peer = toRemotePeerInfo(discNode.record) let peer = toRemotePeerInfo(discNode.record)
if not peer.isOk(): if not peer.isOk():
@ -145,11 +152,11 @@ proc setConnectedPeersMetrics(discoveredNodes: seq[Node],
# crawls the network discovering peers and trying to connect to them # crawls the network discovering peers and trying to connect to them
# metrics are processed and exposed # metrics are processed and exposed
proc crawlNetwork(node: WakuNode, proc crawlNetwork(node: WakuNode,
restClient: RestClientRef,
conf: NetworkMonitorConf, conf: NetworkMonitorConf,
allPeersRef: CustomPeersTableRef) {.async.} = allPeersRef: CustomPeersTableRef) {.async.} =
let crawlInterval = conf.refreshInterval * 1000 let crawlInterval = conf.refreshInterval * 1000
let client = newHttpClient()
while true: while true:
# discover new random nodes # discover new random nodes
let discoveredNodes = await node.wakuDiscv5.protocol.queryRandom() let discoveredNodes = await node.wakuDiscv5.protocol.queryRandom()
@ -163,7 +170,7 @@ proc crawlNetwork(node: WakuNode,
# tries to connect to all newly discovered nodes # tries to connect to all newly discovered nodes
# and populates metrics related to peers we could connect # and populates metrics related to peers we could connect
# note random discovered nodes can be already known # note random discovered nodes can be already known
await setConnectedPeersMetrics(discoveredNodes, node, conf.timeout, client, allPeersRef) await setConnectedPeersMetrics(discoveredNodes, node, conf.timeout, restClient, allPeersRef)
let totalNodes = flatNodes.len let totalNodes = flatNodes.len
let seenNodes = flatNodes.countIt(it.seen) let seenNodes = flatNodes.countIt(it.seen)
@ -281,6 +288,15 @@ when isMainModule:
let res = startRestApiServer(conf, allPeersInfo, msgPerContentTopic) let res = startRestApiServer(conf, allPeersInfo, msgPerContentTopic)
if res.isErr(): if res.isErr():
error "could not start rest api server", err=res.error error "could not start rest api server", err=res.error
quit(1)
# create a rest client
let clientRest = RestClientRef.new(url="http://ip-api.com",
connectTimeout=ctime.seconds(2))
if clientRest.isErr():
error "could not start rest api client", err=res.error
quit(1)
let restClient = clientRest.get()
# start waku node # start waku node
let nodeRes = initAndStartNode(conf) let nodeRes = initAndStartNode(conf)
@ -297,6 +313,6 @@ when isMainModule:
# spawn the routine that crawls the network # spawn the routine that crawls the network
# TODO: split into 3 routines (discovery, connections, ip2location) # TODO: split into 3 routines (discovery, connections, ip2location)
asyncSpawn crawlNetwork(node, conf, allPeersInfo) asyncSpawn crawlNetwork(node, restClient, conf, allPeersInfo)
runForever() runForever()

View File

@ -4,14 +4,16 @@ else:
{.push raises: [].} {.push raises: [].}
import import
std/[json,httpclient], std/json,
stew/results,
stew/shims/net,
chronicles, chronicles,
chronicles/topics_registry, chronicles/topics_registry,
chronos, chronos,
stew/results presto/[client,common]
type type
NodeLocation = object NodeLocation* = object
country*: string country*: string
city*: string city*: string
lat*: string lat*: string
@ -24,30 +26,28 @@ proc flatten*[T](a: seq[seq[T]]): seq[T] =
aFlat &= subseq aFlat &= subseq
return aFlat return aFlat
# using an external api retrieves some data associated with the ip proc decodeBytes*(t: typedesc[NodeLocation], value: openArray[byte],
# TODO: use a cache contentType: Opt[ContentTypeData]): RestResult[NodeLocation] =
# TODO: use nim-presto's HTTP asynchronous client var res: string
proc ipToLocation*(ip: string, if len(value) > 0:
client: Httpclient): res = newString(len(value))
Future[Result[NodeLocation, string]] {.async.} = copyMem(addr res[0], unsafeAddr value[0], len(value))
# naive mechanism to avoid hitting the rate limit
# IP-API endpoints are now limited to 45 HTTP requests per minute
await sleepAsync(1400)
try: try:
let content = client.getContent("http://ip-api.com/json/" & ip) let jsonContent = parseJson(res)
let jsonContent = parseJson(content)
if $jsonContent["status"].getStr() != "success": if $jsonContent["status"].getStr() != "success":
error "query failed", result=jsonContent error "query failed", result=jsonContent
return err("query failed: " & $jsonContent) return err("query failed")
return ok(NodeLocation( return ok(NodeLocation(
country: jsonContent["country"].getStr(), country: jsonContent["country"].getStr(),
city: jsonContent["city"].getStr(), city: jsonContent["city"].getStr(),
lat: jsonContent["lat"].getStr(), lat: $jsonContent["lat"].getFloat(),
long: jsonContent["lon"].getStr(), long: $jsonContent["lon"].getFloat(),
isp: jsonContent["isp"].getStr() isp: jsonContent["isp"].getStr()
)) ))
except: except:
error "failed to get the location for IP", ip=ip, error=getCurrentExceptionMsg() return err("failed to get the location: " & getCurrentExceptionMsg())
return err("failed to get the location for IP '" & ip & "':" & getCurrentExceptionMsg())
proc encodeString*(value: string): RestResult[string] =
ok(value)
proc ipToLocation*(ip: string): RestResponse[NodeLocation] {.rest, endpoint: "json/{ip}", meth: MethodGet.}