mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-08-05 19:33:13 +00:00
6837ae0c1f
* bump nim-libp2p pin to v2.0.0 tag * bump json_rpc to v0.6.1, lsquic to v0.5.1, boringssl to v0.0.8 (latest tags) * add libp2p_mix dep; repoint libp2p/protocols/mix -> libp2p_mix * pin nimble.lock: websock / protobuf_serialization / npeg / jwt * Makefile: add -d:libp2p_quic_support * regenerate nix/deps.nix (adds libp2p_mix, refreshes pins) * migrate rng ref HmacDrbgContext -> libp2p Rng across prod/channels/tests (interface-only; same DRBG) * waku_switch: TransportConfig factory; unified 2.0.0 connection limits (withMaxInOut, withMaxConnections); local MaxConnections * waku_relay/rendezvous/discv5/kademlia: v2.0.0 API (rng, config, ServiceDiscovery rename) * call Service.setup() on post-build switch services (2.0.0 split setup/start) * drop libp2p/utils/semaphore -> chronos AsyncSemaphore * add logos_delivery/waku/compat/option_valueor shim (Option[T] valueOr/withValue, dropped upstream) * add std/options where a transitive re-export was removed * add newStandardSwitch shim (libp2p removed it in 2.0.0); mounts yamux+mplex to match prod muxer * PeerId.random(rng); common.rng()/crypto.newRng(); hoist shared rng (instantiation cleanup) * update expectations for 2.0.0 defaults: DEFAULT_PROTOCOLS += /ipfs/id/push/1.0.0; agent "nim-libp2p" * drop relay reboot/reconnect test (asserted a Switch restart capability that is simply not supported) * fix up a few tests that were flaking on MacOS (libp2p upgrade may have exposed these)
102 lines
3.3 KiB
Nim
102 lines
3.3 KiB
Nim
import logos_delivery/waku/compat/option_valueor
|
|
{.push raises: [].}
|
|
|
|
import chronicles, chronos, metrics, metrics/chronos_httpserver
|
|
import
|
|
logos_delivery/waku/
|
|
[net/auto_port, waku_rln_relay/protocol_metrics as rln_metrics, utils/collector],
|
|
./peer_manager,
|
|
./node_telemetry,
|
|
./waku_node
|
|
|
|
const LogInterval = 10.minutes
|
|
|
|
logScope:
|
|
topics = "waku node metrics"
|
|
|
|
type MetricsServerConf* = object
|
|
httpAddress*: IpAddress
|
|
httpPort*: Port
|
|
logging*: bool
|
|
|
|
proc startMetricsLog*() =
|
|
var logMetrics: CallbackFunc
|
|
|
|
var cumulativeErrors = 0.float64
|
|
var cumulativeConns = 0.float64
|
|
|
|
let logRlnMetrics = getRlnMetricsLogger()
|
|
|
|
logMetrics = CallbackFunc(
|
|
proc(udata: pointer) {.gcsafe.} =
|
|
# TODO: libp2p_pubsub_peers is not public, so we need to make this either
|
|
# public in libp2p or do our own peer counting after all.
|
|
|
|
# track cumulative values
|
|
let freshErrorCount = parseAndAccumulate(waku_node_errors, cumulativeErrors)
|
|
let freshConnCount =
|
|
parseAndAccumulate(waku_node_conns_initiated, cumulativeConns)
|
|
|
|
let totalMessages = collectorAsF64(waku_node_messages)
|
|
let storePeers = collectorAsF64(waku_store_peers)
|
|
let pxPeers = collectorAsF64(waku_px_peers)
|
|
let lightpushPeers = collectorAsF64(waku_lightpush_peers)
|
|
let filterPeers = collectorAsF64(waku_filter_peers)
|
|
|
|
info "Total connections initiated", count = $freshConnCount
|
|
info "Total messages", count = totalMessages
|
|
info "Total store peers", count = storePeers
|
|
info "Total peer exchange peers", count = pxPeers
|
|
info "Total lightpush peers", count = lightpushPeers
|
|
info "Total filter peers", count = filterPeers
|
|
info "Total errors", count = $freshErrorCount
|
|
|
|
# Start protocol specific metrics logging
|
|
logRlnMetrics()
|
|
|
|
discard setTimer(Moment.fromNow(LogInterval), logMetrics)
|
|
)
|
|
|
|
discard setTimer(Moment.fromNow(LogInterval), logMetrics)
|
|
|
|
type StartedMetricsServer* = tuple[server: MetricsHttpServerRef, port: Port]
|
|
|
|
proc startMetricsServer(
|
|
serverIp: IpAddress, serverPort: Port
|
|
): Future[Result[StartedMetricsServer, string]] {.async.} =
|
|
proc attempt(
|
|
port: Port
|
|
): Future[Result[StartedMetricsServer, string]] {.async: (raises: []).} =
|
|
info "Starting metrics HTTP server", serverIp = $serverIp, serverPort = $port
|
|
|
|
let server = MetricsHttpServerRef.new($serverIp, port).valueOr:
|
|
return err("fail to start service metrics server, attempt:" & $error)
|
|
|
|
try:
|
|
await server.start()
|
|
except CatchableError:
|
|
return
|
|
err("exception while startMetricsServer, attempt: " & getCurrentExceptionMsg())
|
|
|
|
info "Metrics HTTP server started", serverIp = $serverIp, serverPort = $port
|
|
return ok((server: server, port: port))
|
|
|
|
let started = (await tryWithAutoPort[StartedMetricsServer](serverPort, attempt)).valueOr:
|
|
return err("metrics HTTP server start failed: " & error)
|
|
return ok(started)
|
|
|
|
proc startMetricsServerAndLogging*(
|
|
conf: MetricsServerConf, portsShift: uint16
|
|
): Future[Result[StartedMetricsServer, string]] {.async.} =
|
|
let started = (
|
|
await (
|
|
startMetricsServer(conf.httpAddress, Port(conf.httpPort.uint16 + portsShift))
|
|
)
|
|
).valueOr:
|
|
return err("Starting metrics server failed. Continuing in current state:" & $error)
|
|
|
|
if conf.logging:
|
|
startMetricsLog()
|
|
|
|
return ok(started)
|