status-im/nimbus-eth2
2
0
Fork 0
mirror of https://github.com/status-im/nimbus-eth2.git synced 2025-02-27 13:40:43 +00:00
Code Issues Projects Releases Wiki Activity

add aggregated attestation tracing to logtrace and enable it in Jenkins (#2766)

Browse Source 
* add aggregated attestation tracing to logtrace and enable it in Jenkins CI

* use a slightly less cryptic acronym than aasr

* mostly, nimbus and the eth2 spec use aggregate attestation, not aggregated attestation
This commit is contained in:
tersec 2021-08-06 21:25:30 +00:00 committed by GitHub
parent 780a9cfd42
commit 61825f4979
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 167 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Jenkinsfile vendored
View File

@ -57,12 +57,12 @@ def runStages() {
// EXECUTOR_NUMBER will be 0 or 1, since we have 2 executors per Jenkins node // EXECUTOR_NUMBER will be 0 or 1, since we have 2 executors per Jenkins node
sh """#!/bin/bash sh """#!/bin/bash
set -e set -e
./scripts/launch_local_testnet.sh --preset minimal --nodes 4 --stop-at-epoch 5 --log-level DEBUG --disable-htop \ ./scripts/launch_local_testnet.sh --preset minimal --nodes 4 --stop-at-epoch 5 --log-level DEBUG --disable-htop --enable-logtrace \
--data-dir local_testnet0_data --base-port \$(( 9000 + EXECUTOR_NUMBER * 100 )) --base-rpc-port \ --data-dir local_testnet0_data --base-port \$(( 9000 + EXECUTOR_NUMBER * 100 )) --base-rpc-port \
\$(( 7000 + EXECUTOR_NUMBER * 100 )) --base-metrics-port \$(( 8008 + EXECUTOR_NUMBER * 100 )) --timeout 600 \ \$(( 7000 + EXECUTOR_NUMBER * 100 )) --base-metrics-port \$(( 8008 + EXECUTOR_NUMBER * 100 )) --timeout 600 \
--kill-old-processes \ --kill-old-processes \
-- --verify-finalization --discv5:no -- --verify-finalization --discv5:no
./scripts/launch_local_testnet.sh --nodes 4 --stop-at-epoch 5 --log-level DEBUG --disable-htop \ ./scripts/launch_local_testnet.sh --nodes 4 --stop-at-epoch 5 --log-level DEBUG --disable-htop --enable-logtrace \
--data-dir local_testnet1_data --base-port \$(( 9000 + EXECUTOR_NUMBER * 100 )) --base-rpc-port \ --data-dir local_testnet1_data --base-port \$(( 9000 + EXECUTOR_NUMBER * 100 )) --base-rpc-port \
\$(( 7000 + EXECUTOR_NUMBER * 100 )) --base-metrics-port \$(( 8008 + EXECUTOR_NUMBER * 100 )) --timeout 2400 \ \$(( 7000 + EXECUTOR_NUMBER * 100 )) --base-metrics-port \$(( 8008 + EXECUTOR_NUMBER * 100 )) --timeout 2400 \
--kill-old-processes \ --kill-old-processes \

257
ncli/logtrace.nim
View File

@ -1,5 +1,5 @@
# beacon_chain # beacon_chain
# Copyright (c) 2018-2020 Status Research & Development GmbH # Copyright (c) 2018-2021 Status Research & Development GmbH
# Licensed and distributed under either of # Licensed and distributed under either of
# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT). # * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0). # * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
@ -9,154 +9,159 @@ import confutils, json, times, streams, os, strutils, options, chronicles,
import json_serialization import json_serialization
const const
LogTraceName* = "Beacon-Chain LogTrace Tool" LogTraceName = "Beacon-Chain LogTrace Tool"
LogTraceMajor*: int = 0 LogTraceMajor: int = 0
LogTraceMinor*: int = 0 LogTraceMinor: int = 0
LogTracePatch*: int = 4 LogTracePatch: int = 4
LogTraceVersion* = $LogTraceMajor & "." & $LogTraceMinor & "." & LogTraceVersion = $LogTraceMajor & "." & $LogTraceMinor & "." &
$LogTracePatch $LogTracePatch
LogTraceCopyright* = "Copyright(C) 2020" & LogTraceCopyright = "Copyright(C) 2021" &
" Status Research & Development GmbH" " Status Research & Development GmbH"
LogTraceHeader* = LogTraceName & ", Version " & LogTraceVersion & LogTraceHeader = LogTraceName & ", Version " & LogTraceVersion &
" [" & hostOS & ": " & hostCPU & "]\r\n" & " [" & hostOS & ": " & hostCPU & "]\r\n" &
LogTraceCopyright & "\r\n" LogTraceCopyright & "\r\n"
type type
StartUpCommand* {.pure.} = enum StartUpCommand {.pure.} = enum
pubsub, asl, asr, lat pubsub, asl, asr, aggasr, lat
LogTraceConf* = object LogTraceConf = object
logFiles* {. logFiles {.
desc: "Specifies one or more log files", desc: "Specifies one or more log files",
abbr: "f", abbr: "f",
name: "log-file" }: seq[string] name: "log-file" }: seq[string]
simDir* {. simDir {.
desc: "Specifies path to eth2_network_simulation directory", desc: "Specifies path to eth2_network_simulation directory",
name: "sim-dir", name: "sim-dir",
defaultValue: "" }: string defaultValue: "" }: string
netDir* {. netDir {.
desc: "Specifies path to network build directory", desc: "Specifies path to network build directory",
name: "net-dir", name: "net-dir",
defaultValue: "" }: string defaultValue: "" }: string
logDir* {. logDir {.
desc: "Specifies path with bunch of logs", desc: "Specifies path with bunch of logs",
name: "log-dir", name: "log-dir",
defaultValue: "" }: string defaultValue: "" }: string
ignoreSerializationErrors* {. ignoreSerializationErrors {.
desc: "Ignore serialization errors while parsing log files", desc: "Ignore serialization errors while parsing log files",
name: "ignore-errors", name: "ignore-errors",
defaultValue: true }: bool defaultValue: true }: bool
dumpSerializationErrors* {. dumpSerializationErrors {.
desc: "Dump full serialization errors while parsing log files", desc: "Dump full serialization errors while parsing log files",
name: "dump-errors", name: "dump-errors",
defaultValue: false }: bool defaultValue: false }: bool
nodes* {. nodes {.
desc: "Specifies node names which logs will be used", desc: "Specifies node names which logs will be used",
name: "nodes" }: seq[string] name: "nodes" }: seq[string]
allowedLag* {. allowedLag {.
desc: "Allowed latency lag multiplier", desc: "Allowed latency lag multiplier",
name: "lag", name: "lag",
defaultValue: 2.0 }: float defaultValue: 2.0 }: float
case cmd* {.command.}: StartUpCommand case cmd {.command.}: StartUpCommand
of pubsub: of pubsub:
discard discard
of asl: of asl:
discard discard
of asr: of asr:
discard discard
of aggasr:
discard
of lat: of lat:
discard discard
GossipDirection* = enum GossipDirection = enum
None, Incoming, Outgoing None, Incoming, Outgoing
NodeDirectory* = object NodeDirectory = object
name*: string name: string
path*: string path: string
logs*: seq[string] logs: seq[string]
LogMessage* = object of RootObj LogMessage = object of RootObj
level* {.serializedFieldName: "lvl" .}: string level {.serializedFieldName: "lvl" .}: string
timestamp* {.serializedFieldName: "ts" .}: DateTime timestamp {.serializedFieldName: "ts" .}: DateTime
msg*: string msg: string
topics*: string topics: string
tid*: int tid: int
SlotStartMessage* = object of LogMessage SlotStartMessage = object of LogMessage
beaconTime*: uint64 beaconTime: uint64
finalizedEpoch*: uint64 finalizedEpoch: uint64
finalizedRoot*: string finalizedRoot: string
finalizedSlot*: uint64 finalizedSlot: uint64
headEpoch*: uint64 headEpoch: uint64
headRoot*: string headRoot: string
headSlot*: uint64 headSlot: uint64
lastSlot*: uint64 lastSlot: uint64
peers*: uint64 peers: uint64
scheduledSlot*: uint64 scheduledSlot: uint64
AttestationDataObject* = object AttestationDataObject = object
slot*: uint64 slot: uint64
index*: uint64 index: uint64
beaconBlockRoot* {.serializedFieldName: "beacon_block_root".}: string beaconBlockRoot {.serializedFieldName: "beacon_block_root".}: string
sourceEpoch* {.serializedFieldName: "source_epoch".}: uint64 sourceEpoch {.serializedFieldName: "source_epoch".}: uint64
sourceRoot* {.serializedFieldName: "source_root".}: string sourceRoot {.serializedFieldName: "source_root".}: string
targetEpoch* {.serializedFieldName: "target_epoch".}: uint64 targetEpoch {.serializedFieldName: "target_epoch".}: uint64
targetRoot* {.serializedFieldName: "target_root".}: string targetRoot {.serializedFieldName: "target_root".}: string
AttestationObject* = object AttestationObject = object
aggregationBits* {.serializedFieldName: "aggregation_bits".}: string aggregationBits {.serializedFieldName: "aggregation_bits".}: string
data*: AttestationDataObject data: AttestationDataObject
signature*: string signature: string
AttestationSentMessage* = object of LogMessage AttestationSentMessage = object of LogMessage
attestation*: AttestationObject attestation: AttestationObject
indexInCommittee*: uint64 indexInCommittee: uint64
validator*: string validator: string
AttestationReceivedMessage* = object of LogMessage AttestationReceivedMessage = object of LogMessage
attestation*: AttestationObject attestation: AttestationObject
head*: string head: string
wallSlot*: uint64 wallSlot: uint64
pcs*: string pcs: string
GossipMessage* = object AggregatedAttestationSentMessage = object of LogMessage
kind*: GossipDirection attestation: AttestationObject
id*: string validator: string
datetime*: DateTime signature: string
processed*: bool aggregationSlot: uint64
SaMessageType* {.pure.} = enum AggregatedAttestationReceivedMessage = object of LogMessage
aggregate: AttestationObject
wallSlot: uint64
signature: string
GossipMessage = object
kind: GossipDirection
id: string
datetime: DateTime
processed: bool
SaMessageType {.pure.} = enum
AttestationSent, SlotStart AttestationSent, SlotStart
SRMessageType* {.pure.} = enum SlotAttMessage = object
AttestationSent, AttestationReceived case kind: SaMessageType
SlotAttMessage* = object
case kind*: SaMessageType
of SaMessageType.AttestationSent: of SaMessageType.AttestationSent:
asmsg*: AttestationSentMessage asmsg: AttestationSentMessage
of SaMessageType.SlotStart: of SaMessageType.SlotStart:
ssmsg*: SlotStartMessage ssmsg: SlotStartMessage
SRAttMessage* = object SRANode = object
case kind*: SRMessageType directory: NodeDirectory
of SRMessageType.AttestationSent: sends: seq[AttestationSentMessage]
asmsg*: AttestationSentMessage recvs: TableRef[string, AttestationReceivedMessage]
of SRMessageType.AttestationReceived: aggSends: seq[AggregatedAttestationSentMessage]
armsg*: AttestationReceivedMessage aggRecvs: TableRef[string, AggregatedAttestationReceivedMessage]
SRANode* = object
directory*: NodeDirectory
sends*: seq[AttestationSentMessage]
recvs*: TableRef[string, AttestationReceivedMessage]
proc readValue*(reader: var JsonReader, value: var DateTime) = proc readValue*(reader: var JsonReader, value: var DateTime) =
let s = reader.readValue(string) let s = reader.readValue(string)
@ -273,6 +278,14 @@ proc readLogFileForASRMessages(file: string, srnode: var SRANode,
let rm = Json.decode(line, AttestationReceivedMessage, let rm = Json.decode(line, AttestationReceivedMessage,
allowUnknownFields = true) allowUnknownFields = true)
discard srnode.recvs.hasKeyOrPut(rm.attestation.signature, rm) discard srnode.recvs.hasKeyOrPut(rm.attestation.signature, rm)
elif m.msg == "Aggregate received":
let rm = Json.decode(line, AggregatedAttestationReceivedMessage,
allowUnknownFields = true)
discard srnode.aggRecvs.hasKeyOrPut(rm.signature, rm)
elif m.msg == "Aggregated attestation sent":
let sm = Json.decode(line, AggregatedAttestationSentMessage,
allowUnknownFields = true)
srnode.aggSends.add(sm)
if counter mod 10_000 == 0: if counter mod 10_000 == 0:
info "Processing file", file = extractFilename(file), info "Processing file", file = extractFilename(file),
@ -410,7 +423,7 @@ proc runPubsub(logConf: LogTraceConf, logFiles: seq[string]) =
var logs = newSeq[tuple[name: string, data: seq[GossipMessage]]]() var logs = newSeq[tuple[name: string, data: seq[GossipMessage]]]()
if len(logFiles) < 2: if len(logFiles) < 2:
error "Number of log files are not enough to process pubsub messages", error "Number of log files insufficient to process pubsub messages",
logs_count = len(logFiles) logs_count = len(logFiles)
quit(1) quit(1)
@ -450,7 +463,7 @@ proc runPubsub(logConf: LogTraceConf, logFiles: seq[string]) =
proc runAttSend(logConf: LogTraceConf, logFiles: seq[string]) = proc runAttSend(logConf: LogTraceConf, logFiles: seq[string]) =
info "Check for late `attestation sent` messages" info "Check for late `attestation sent` messages"
if len(logFiles) < 1: if len(logFiles) < 1:
error "Number of log files are not enough to process pubsub messages", error "Number of log files insufficient to process pubsub messages",
logs_count = len(logFiles) logs_count = len(logFiles)
quit(1) quit(1)
@ -491,9 +504,9 @@ proc toSimple*(s: seq[string]): string =
result = "[" & s.mapIt("'" & it & "'").join(", ") & "]" result = "[" & s.mapIt("'" & it & "'").join(", ") & "]"
proc runAttSendReceive(logConf: LogTraceConf, nodes: seq[NodeDirectory]) = proc runAttSendReceive(logConf: LogTraceConf, nodes: seq[NodeDirectory]) =
info "Check for attestations send/receive messages" info "Check for attestation sent/received messages"
if len(nodes) < 2: if len(nodes) < 2:
error "Number of nodes' log files are not enough", nodes_count = len(nodes) error "Number of nodes' log files insufficient", nodes_count = len(nodes)
quit(1) quit(1)
var srnodes = newSeq[SRANode]() var srnodes = newSeq[SRANode]()
@ -501,7 +514,9 @@ proc runAttSendReceive(logConf: LogTraceConf, nodes: seq[NodeDirectory]) =
var srnode = SRANode( var srnode = SRANode(
directory: node, directory: node,
sends: newSeq[AttestationSentMessage](), sends: newSeq[AttestationSentMessage](),
recvs: newTable[string, AttestationReceivedMessage]() recvs: newTable[string, AttestationReceivedMessage](),
aggSends: newSeq[AggregatedAttestationSentMessage](),
aggRecvs: newTable[string, AggregatedAttestationReceivedMessage]()
) )
info "Processing node", node = node.name info "Processing node", node = node.name
for logfile in node.logs: for logfile in node.logs:
@ -513,7 +528,7 @@ proc runAttSendReceive(logConf: LogTraceConf, nodes: seq[NodeDirectory]) =
srnodes.add(srnode) srnodes.add(srnode)
if len(nodes) < 2: if len(nodes) < 2:
error "Number of nodes' log files are not enough", nodes_count = len(nodes) error "Number of nodes' log files insufficient", nodes_count = len(nodes)
quit(1) quit(1)
for i in 0 ..< len(srnodes): for i in 0 ..< len(srnodes):
@ -539,11 +554,63 @@ proc runAttSendReceive(logConf: LogTraceConf, nodes: seq[NodeDirectory]) =
successful_broadcasts = success, failed_broadcasts = failed, successful_broadcasts = success, failed_broadcasts = failed,
total_broadcasts = len(srnodes[i].sends) total_broadcasts = len(srnodes[i].sends)
proc runAggAttSendReceive(logConf: LogTraceConf, nodes: seq[NodeDirectory]) =
info "Check for aggregate attestation sent/received messages"
if len(nodes) < 2:
error "Number of nodes' log files insufficient", nodes_count = len(nodes)
quit(1)
var srnodes = newSeq[SRANode]()
for node in nodes:
var srnode = SRANode(
directory: node,
sends: newSeq[AttestationSentMessage](),
recvs: newTable[string, AttestationReceivedMessage](),
aggSends: newSeq[AggregatedAttestationSentMessage](),
aggRecvs: newTable[string, AggregatedAttestationReceivedMessage]()
)
info "Processing node", node = node.name
for logfile in node.logs:
let path = node.path & DirSep & logfile
info "Processing node's logfile", node = node.name, logfile = path
readLogFileForASRMessages(path, srnode,
logConf.ignoreSerializationErrors,
logConf.dumpSerializationErrors)
srnodes.add(srnode)
if len(nodes) < 2:
error "Number of nodes' log files insufficient", nodes_count = len(nodes)
quit(1)
for i in 0 ..< len(srnodes):
var success = 0
var failed = 0
for item in srnodes[i].aggSends:
var k = (i + 1) mod len(srnodes)
var misses = newSeq[string]()
while k != i:
if item.signature notin srnodes[k].aggRecvs:
misses.add(srnodes[k].directory.name)
k = (k + 1) mod len(srnodes)
if len(misses) == 0:
inc(success)
else:
inc(failed)
info "Aggregate attestation was not received",
sender = srnodes[i].directory.name,
signature = item.signature,
receivers = misses.toSimple(), send_stamp = item.timestamp
info "Statistics for sender node", sender = srnodes[i].directory.name,
successful_broadcasts = success, failed_broadcasts = failed,
total_broadcasts = len(srnodes[i].aggSends)
proc runLatencyCheck(logConf: LogTraceConf, logFiles: seq[string], proc runLatencyCheck(logConf: LogTraceConf, logFiles: seq[string],
nodes: seq[NodeDirectory]) = nodes: seq[NodeDirectory]) =
info "Check for async responsiveness" info "Check for async responsiveness"
if len(nodes) == 0 and len(logFiles) == 0: if len(nodes) == 0 and len(logFiles) == 0:
error "Number of log files are not enough", nodes_count = len(nodes) error "Number of log files insufficient", nodes_count = len(nodes)
quit(1) quit(1)
let allowedTime = int64(float(initDuration(seconds = 1).inMilliseconds()) * let allowedTime = int64(float(initDuration(seconds = 1).inMilliseconds()) *
@ -617,6 +684,8 @@ proc run(conf: LogTraceConf) =
runAttSend(conf, logFiles) runAttSend(conf, logFiles)
of StartUpCommand.asr: of StartUpCommand.asr:
runAttSendReceive(conf, logNodes) runAttSendReceive(conf, logNodes)
of StartUpCommand.aggasr:
runAggAttSendReceive(conf, logNodes)
of StartUpCommand.lat: of StartUpCommand.lat:
runLatencyCheck(conf, logFiles, logNodes) runLatencyCheck(conf, logFiles, logNodes)

4
scripts/launch_local_testnet.sh
View File

@ -69,7 +69,7 @@ CI run: $(basename "$0") --disable-htop -- --verify-finalization
--base-metrics-port bootstrap node's metrics server port (default: ${BASE_METRICS_PORT}) --base-metrics-port bootstrap node's metrics server port (default: ${BASE_METRICS_PORT})
--disable-htop don't use "htop" to see the nimbus_beacon_node processes --disable-htop don't use "htop" to see the nimbus_beacon_node processes
--disable-vc don't use validator client binaries for validators (by default validators are split 50/50 between beacon nodes and validator clients) --disable-vc don't use validator client binaries for validators (by default validators are split 50/50 between beacon nodes and validator clients)
--enable-logtrace display logtrace asr analysis --enable-logtrace display logtrace aggasr analysis
--log-level set the log level (default: ${LOG_LEVEL}) --log-level set the log level (default: ${LOG_LEVEL})
--reuse-existing-data-dir instead of deleting and recreating the data dir, keep it and reuse everything we can from it --reuse-existing-data-dir instead of deleting and recreating the data dir, keep it and reuse everything we can from it
--timeout timeout in seconds (default: ${TIMEOUT_DURATION} - no timeout) --timeout timeout in seconds (default: ${TIMEOUT_DURATION} - no timeout)
@ -351,7 +351,7 @@ dump_logs() {
dump_logtrace() { dump_logtrace() {
if [[ "$ENABLE_LOGTRACE" == "1" ]]; then if [[ "$ENABLE_LOGTRACE" == "1" ]]; then
find "${DATA_DIR}" -maxdepth 1 -type f -regex '.*/log[0-9]+.txt' | sed -e"s/${DATA_DIR}\//--nodes=/" | sort | xargs ./build/logtrace asr --log-dir="${DATA_DIR}" || true find "${DATA_DIR}" -maxdepth 1 -type f -regex '.*/log[0-9]+.txt' | sed -e"s/${DATA_DIR}\//--nodes=/" | sort | xargs ./build/logtrace aggasr --log-dir="${DATA_DIR}" || true
fi fi
} }