forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 18:32:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

dashboard, p2p: fix peer limiting bug, properly visualize traffic data diff

Browse source
This commit is contained in:
Kurkó Mihály 2018-10-17 15:09:53 +03:00
parent 8563484836
commit ad1b55ba38
5 changed files with 601 additions and 477 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

620
dashboard/assets.go
View file

File diff suppressed because it is too large Load diff

127
dashboard/assets/components/Network.jsx
View file

@ -36,20 +36,45 @@ export const inserter = (sampleLimit: number) => (update: NetworkType, prev: Net
}
if (Array.isArray(update.diff)) {
update.diff.forEach((event: PeerEvent) => {
if (event.removeIP) {
if (event.removeID && prev.peers.bundles[event.removeIP]) {
delete prev.peers.bundles[event.removeIP].knownPeers[event.removeID];
}
delete prev.peers.bundles[event.removeIP];
return;
}
if (!event.ip) {
console.error('Peer event without IP', event);
return;
}
switch (event.remove) {
case 'bundle':
delete prev.peers.bundles[event.ip];
return;
case 'known': {
if (!event.id) {
console.error('Remove known peer event without ID', event.ip);
return;
}
const bundle = prev.peers.bundles[event.ip];
if (!bundle || !bundle.knownPeers || !bundle.knownPeers[event.id]) {
console.error('No known peer to remove', event.ip, event.id);
return;
}
delete bundle.knownPeers[event.id];
return;
}
case 'unknown': {
const bundle = prev.peers.bundles[event.ip];
if (!bundle || !Array.isArray(bundle.unknownPeers) || bundle.unknownPeers.length < 1) {
console.error('No unknown peer to remove', event.ip);
return;
}
bundle.unknownPeers.splice(0, 1);
return;
}
}
if (!prev.peers.bundles[event.ip]) {
prev.peers.bundles[event.ip] = {
location: {},
location: {
country: '',
city: '',
latitude: 0,
longitude: 0,
},
knownPeers: {},
unknownPeers: [],
};
@ -57,6 +82,7 @@ export const inserter = (sampleLimit: number) => (update: NetworkType, prev: Net
const bundle = prev.peers.bundles[event.ip];
if (event.location) {
bundle.location = event.location;
return;
}
if (!event.id) {
bundle.unknownPeers.push({
@ -85,16 +111,6 @@ export const inserter = (sampleLimit: number) => (update: NetworkType, prev: Net
console.error('Different traffic sample length', event);
return;
}
if (peer.ingress.length > 0) {
if (peer.ingress[peer.ingress.length - 1].value < event.ingress[0].value) {
event.ingress[0].value -= peer.ingress[peer.ingress.length - 1].value;
event.egress[0].value -= peer.egress[peer.egress.length - 1].value;
}
}
for (let i = 1; i < event.ingress.length; i++) {
event.ingress[i].value -= event.ingress[i - 1].value;
event.egress[i].value -= event.egress[i - 1].value;
}
peer.ingress.splice(peer.ingress.length, 0, ...event.ingress);
peer.egress.splice(peer.egress.length, 0, ...event.egress);
if (peer.ingress.length > sampleLimit) {
@ -103,7 +119,6 @@ export const inserter = (sampleLimit: number) => (update: NetworkType, prev: Net
if (peer.egress.length > sampleLimit) {
peer.egress.splice(0, peer.egress.length - sampleLimit);
}
// console.log(event.ingress, prev.peers.bundles[event.ip].knownPeers[event.id].ingress);
}
});
}
@ -138,12 +153,12 @@ class Network extends Component<Props, State> {
render() {
return (
<div>
<Table>
<TableHead>
<TableRow>
<TableCell>IP</TableCell>
<TableCell>Location</TableCell>
<TableCell>Unknown</TableCell>
<TableCell>Node ID</TableCell>
<TableCell>Traffic</TableCell>
<TableCell>Connected</TableCell>
@ -151,8 +166,12 @@ class Network extends Component<Props, State> {
</TableRow>
</TableHead>
<TableBody>
{Object.entries(this.props.content.peers.bundles).map(([ip, bundle]) => (
<TableRow key={ip}>
{Object.entries(this.props.content.peers.bundles).map(([ip, bundle]) => {
if (!bundle.knownPeers || Object.keys(bundle.knownPeers).length < 1) {
return null;
}
return (
<TableRow key={`known${ip}`}>
<TableCell>{ip}</TableCell>
<TableCell>
{bundle.location ? (() => {
@ -161,46 +180,78 @@ class Network extends Component<Props, State> {
})() : ''}
</TableCell>
<TableCell>
{bundle.unknownPeers && Object.values(bundle.unknownPeers).map(peer => peer.connected && peer.disconnected && `${this.formatTime(peer.connected)}~${this.formatTime(peer.disconnected)}`).join(', ')}
{Object.keys(bundle.knownPeers).map(id => id.substring(0, 10)).join(' ')}
</TableCell>
<TableCell>
{bundle.knownPeers && Object.keys(bundle.knownPeers).map(id => id.substring(0, 10)).join(' ')}
</TableCell>
<TableCell>
{bundle.knownPeers && Object.values(bundle.knownPeers).map(({ingress, egress}) => (
{Object.values(bundle.knownPeers).map(({ingress, egress}) => (
<div>
<AreaChart
width={200} height={50}
width={300} height={50}
syncId={'footerSyncId'}
data={ingress.map(({value}) => ({ingress: value || 0}))}
data={egress.map(({value}) => ({egress: value || 0}))}
margin={{top: 5, right: 5, bottom: 0, left: 5}}
>
<Tooltip cursor={false} content={<CustomTooltip tooltip={bytePlotter('Upload')} />} />
<Area isAnimationActive={false} type='monotone' dataKey='ingress' stroke='#8884d8' fill='#8884d8' />
<Tooltip cursor={false} content={<CustomTooltip tooltip={bytePlotter('Download')} />} />
<Area isAnimationActive={false} type='monotone' dataKey='egress' stroke='#8884d8' fill='#8884d8' />
</AreaChart>
<AreaChart
width={200} height={50}
width={300} height={50}
syncId={'footerSyncId'}
data={egress.map(({value}) => ({egress: -value || 0}))}
data={ingress.map(({value}) => ({ingress: -value || 0}))}
margin={{top: 0, right: 5, bottom: 5, left: 5}}
>
<Tooltip cursor={false} content={<CustomTooltip tooltip={bytePlotter('Download', multiplier(-1))} />} />
<Area isAnimationActive={false} type='monotone' dataKey='egress' stroke='#82ca9d' fill='#82ca9d' />
<Tooltip cursor={false} content={<CustomTooltip tooltip={bytePlotter('Upload', multiplier(-1))} />} />
<Area isAnimationActive={false} type='monotone' dataKey='ingress' stroke='#82ca9d' fill='#82ca9d' />
</AreaChart>
</div>
))}
</TableCell>
<TableCell>
{bundle.knownPeers && Object.values(bundle.knownPeers).map(peer => peer.connected && peer.connected.map(time => this.formatTime(time)).join(' ')).join(', ')}
{Object.values(bundle.knownPeers).map(peer => peer.connected && peer.connected.map(time => this.formatTime(time)).join(' ')).join(', ')}
</TableCell>
<TableCell>
{bundle.knownPeers && Object.values(bundle.knownPeers).map(peer => peer.disconnected && peer.disconnected.map(time => this.formatTime(time)).join(' ')).join(', ')}
{Object.values(bundle.knownPeers).map(peer => peer.disconnected && peer.disconnected.map(time => this.formatTime(time)).join(' ')).join(', ')}
</TableCell>
</TableRow>
))}
);
})}
</TableBody>
</Table>
<Table>
<TableHead>
<TableRow>
<TableCell>IP</TableCell>
<TableCell>Location</TableCell>
<TableCell>Connected</TableCell>
<TableCell>Disconnected</TableCell>
</TableRow>
</TableHead>
<TableBody>
{Object.entries(this.props.content.peers.bundles).map(([ip, bundle]) => {
if (!bundle.unknownPeers || bundle.unknownPeers.length < 1) {
return null;
}
return (
<TableRow key={`unknown${ip}`}>
<TableCell>{ip}</TableCell>
<TableCell>
{bundle.location ? (() => {
const l = bundle.location;
return `${l.country ? l.country : ''}${l.city ? `/${l.city}` : ''} ${l.latitude} ${l.longitude}`;
})() : ''}
</TableCell>
<TableCell>
{Object.values(bundle.unknownPeers).map(peer => peer.connected && `${this.formatTime(peer.connected)}`).join(', ')}
</TableCell>
<TableCell>
{Object.values(bundle.unknownPeers).map(peer => peer.disconnected && `${this.formatTime(peer.disconnected)}`).join(', ')}
</TableCell>
</TableRow>
);
})}
</TableBody>
</Table>
</div>);
}
}

3
dashboard/assets/types/content.jsx
View file

@ -58,8 +58,7 @@ export type Network = {
export type PeerEvent = {
ip: string,
id: string,
removeIP: string,
removeID: string,
remove: string,
location: GeoLocation,
connected: Date,
disconnected: Date,

215
dashboard/peers.go
View file

@ -29,8 +29,8 @@ import (
const (
eventBufferLimit = 128 // Maximum number of events of buffered peer events.
knownPeerLimit = p2p.MeteredPeerLimit // Maximum number of stored peers, which successfully made the handshake.
unknownPeerLimit = p2p.MeteredPeerLimit // Maximum number of stored peers, which failed to make the handshake.
knownPeerLimit = 30 //p2p.MeteredPeerLimit // Maximum number of stored peers, which successfully made the handshake.
unknownPeerLimit = 100 //p2p.MeteredPeerLimit // Maximum number of stored peers, which failed to make the handshake.
// eventLimit is the maximum number of the dashboard's custom peer events,
// that are collected between two metering period and sent to the clients
@ -136,27 +136,16 @@ func (pc *PeerContainer) getOrInitBundle(ip string) (*PeerBundle, []*PeerEvent)
return pc.Bundles[ip], events
}
func (pc *PeerContainer) getOrInitKnownPeer(ip, id string) (*KnownPeer, []*PeerEvent) {
bundle, events := pc.getOrInitBundle(ip)
peer, peerEvents := bundle.getOrInitKnownPeer(ip, id, pc.refresh)
return peer, append(events, peerEvents...)
}
// extendKnown handles the events of the successfully connected peers.
// Returns the events occurring during the extension.
func (pc *PeerContainer) extendKnown(event *PeerEvent) (events []*PeerEvent) {
bundle, bundleInitEvents := pc.getOrInitBundle(event.IP)
events = append(events, bundleInitEvents...)
peer, peerInitEvents := bundle.getOrInitKnownPeer(event.IP, event.ID, pc.refresh)
events = append(events, peerInitEvents...)
if peer.listElement != nil {
// If the peer is already in the list, remove it and reinsert later.
pc.knownPeers.Remove(peer.listElement)
peer.listElement = nil
}
for pc.knownPeers.Len() >= knownPeerLimit {
// While the length of the list is greater than or equal to the limit before
// the insertion, remove the first element from the list and from the map.
if removedPeer, ok := pc.knownPeers.Remove(pc.knownPeers.Front()).(*KnownPeer); ok {
if event := pc.removeKnown(removedPeer.ip, removedPeer.id); event != nil {
events = append(events, event)
}
}
}
func (pc *PeerContainer) extendKnown(event *PeerEvent) []*PeerEvent {
peer, events := pc.getOrInitKnownPeer(event.IP, event.ID)
// Append the connect and the disconnect events to
// the corresponding arrays keeping the limit.
if event.Connected != nil {
@ -171,24 +160,44 @@ func (pc *PeerContainer) extendKnown(event *PeerEvent) (events []*PeerEvent) {
peer.Disconnected = peer.Disconnected[first:]
}
}
if peer.listElement != nil {
if pc.activeSeparator == peer.listElement {
pc.activeSeparator = pc.activeSeparator.Prev()
}
// If the peer is already in the list, remove and reinsert it.
_ = pc.knownPeers.Remove(peer.listElement)
peer.listElement = nil
}
// Insert the peer into the list.
if pc.activeSeparator == nil {
// If there isn't active peer in the list
peer.listElement = pc.knownPeers.PushBack(peer)
pc.activeSeparator = peer.listElement
} else if e := pc.knownPeers.InsertAfter(peer, pc.activeSeparator); e != nil {
// Insert the peer after the last inactive peer, and set it as the separator.
peer.listElement, pc.activeSeparator = e, e
} else {
// Insert the peer after the last inactive peer, then increment the separator.
peer.listElement = pc.knownPeers.InsertAfter(peer, pc.activeSeparator)
pc.activeSeparator = pc.activeSeparator.Next()
log.Warn("Failed to insert known peer", "peer", *peer)
}
for pc.knownPeers.Len() > knownPeerLimit {
// While the length of the list is greater than the limit,
// remove the first element from the list and from the map.
if pc.activeSeparator == pc.knownPeers.Front() {
pc.activeSeparator = nil
}
if removedPeer, ok := pc.knownPeers.Remove(pc.knownPeers.Front()).(*KnownPeer); ok {
events = append(events, pc.removeKnown(removedPeer.ip, removedPeer.id)...)
} else {
log.Warn("Failed to parse the removed peer")
}
}
return events
}
// extendUnknown handles the events of the peers failing before/during the handshake.
// Returns the events occurring during the extension.
func (pc *PeerContainer) extendUnknown(event *PeerEvent) (events []*PeerEvent) {
bundle, initEvents := pc.getOrInitBundle(event.IP)
events = append(events, initEvents...)
func (pc *PeerContainer) extendUnknown(event *PeerEvent) []*PeerEvent {
bundle, events := pc.getOrInitBundle(event.IP)
bundle.UnknownPeers = append(bundle.UnknownPeers, &UnknownPeer{
Connected: *event.Connected,
Disconnected: *event.Disconnected,
@ -198,16 +207,14 @@ func (pc *PeerContainer) extendUnknown(event *PeerEvent) (events []*PeerEvent) {
// While the length of the connection attempt order array is greater
// than the limit, remove the first element from the involved peer's
// array and also from the super array.
if r := pc.removeUnknown(pc.unknownPeers[0]); r != nil {
events = append(events, r)
}
events = append(events, pc.removeUnknown(pc.unknownPeers[0])...)
pc.unknownPeers = pc.unknownPeers[1:]
}
return events
}
// setActive moves the peer denoted by the given IP address and node ID after
// the list's active separator.
// the list's active separator. Takes no effect if the peer doesn't exist.
func (pc *PeerContainer) setActive(ip, id string) {
if bundle, ok := pc.Bundles[ip]; ok {
if peer, ok := bundle.KnownPeers[id]; ok {
@ -215,14 +222,19 @@ func (pc *PeerContainer) setActive(ip, id string) {
// If the peer is already in the list, remove it first.
// Theoretically this should always happen, because all
// the peers are inserted into the list.
pc.knownPeers.Remove(peer.listElement)
if pc.activeSeparator == peer.listElement {
pc.activeSeparator = pc.activeSeparator.Prev()
}
_ = pc.knownPeers.Remove(peer.listElement)
}
if pc.activeSeparator == nil {
// If there isn't active peer yet.
peer.listElement = pc.knownPeers.PushBack(peer)
pc.activeSeparator = peer.listElement
} else if e := pc.knownPeers.InsertAfter(peer, pc.activeSeparator); e != nil {
peer.listElement = e
} else {
peer.listElement = pc.knownPeers.InsertAfter(peer, pc.activeSeparator)
log.Warn("Failed to insert the peer after the separator", "peer", peer)
}
}
}
@ -250,41 +262,53 @@ type PeerBundle struct {
// removeKnown removes the known peer belonging to the
// given IP address and node ID from the peer tree.
func (pc *PeerContainer) removeKnown(ip, id string) (removed *PeerEvent) {
func (pc *PeerContainer) removeKnown(ip, id string) (events []*PeerEvent) {
if bundle, ok := pc.Bundles[ip]; ok {
if _, ok := bundle.KnownPeers[id]; ok {
removed = &PeerEvent{
RemoveID: id,
}
events = append(events, &PeerEvent{
Remove: RemoveKnown,
IP: ip,
ID: id,
})
delete(bundle.KnownPeers, id)
} else {
log.Warn("No peer to remove", ip, id)
}
if len(bundle.KnownPeers) < 1 && len(bundle.UnknownPeers) < 1 {
if removed == nil {
removed = &PeerEvent{
RemoveIP: ip,
}
} else {
removed.RemoveIP = ip
}
events = append(events, &PeerEvent{
Remove: RemoveBundle,
IP: ip,
})
delete(pc.Bundles, ip)
}
} else {
log.Warn("No bundle to remove", ip)
}
return removed
return events
}
// removeUnknown removes the unknown peer belonging to the
// given IP address and node ID from the peer tree.
func (pc *PeerContainer) removeUnknown(ip string) (removed *PeerEvent) {
func (pc *PeerContainer) removeUnknown(ip string) (events []*PeerEvent) {
if bundle, ok := pc.Bundles[ip]; ok {
if len(bundle.UnknownPeers) > 0 {
events = append(events, &PeerEvent{
Remove: RemoveUnknown,
IP: ip,
})
bundle.UnknownPeers = bundle.UnknownPeers[1:]
}
if len(bundle.KnownPeers) < 1 && len(bundle.UnknownPeers) < 1 {
removed = &PeerEvent{RemoveIP: ip}
if len(bundle.UnknownPeers) < 1 && len(bundle.KnownPeers) < 1 {
events = append(events, &PeerEvent{
Remove: RemoveBundle,
IP: ip,
})
delete(pc.Bundles, ip)
}
} else {
log.Warn("No bundle to remove", ip)
}
return removed
return events
}
// getOrInitKnownPeer inserts a new peer into the map, if the peer belonging
@ -300,8 +324,8 @@ func (bundle *PeerBundle) getOrInitKnownPeer(ip, id string, refresh time.Duratio
events = append(events, &PeerEvent{
IP: ip,
ID: id,
Ingress: ingress,
Egress: egress,
Ingress: append([]*ChartEntry{}, ingress...),
Egress: append([]*ChartEntry{}, egress...),
})
bundle.KnownPeers[id] = &KnownPeer{
ip: ip,
@ -333,6 +357,8 @@ type KnownPeer struct {
listElement *list.Element // Pointer to the peer element in the list.
ip, id string // The IP and the ID by which the peer can be accessed in the tree.
prevIngress float64
prevEgress float64
}
// UnknownPeer contains a failed peer connection attempt's attributes.
@ -345,12 +371,19 @@ type UnknownPeer struct {
Disconnected time.Time `json:"disconnected"`
}
type RemovedPeerType string
const (
RemoveKnown RemovedPeerType = "known"
RemoveUnknown RemovedPeerType = "unknown"
RemoveBundle RemovedPeerType = "bundle"
)
// PeerEvent contains the attributes of a peer event.
type PeerEvent struct {
IP string `json:"ip,omitempty"` // IP address of the peer.
ID string `json:"id,omitempty"` // Node ID of the peer.
RemoveIP string `json:"removeIP,omitempty"` // IP address of the peer that is to be removed.
RemoveID string `json:"removeID,omitempty"` // Node ID of the peer that is to be removed.
Remove RemovedPeerType `json:"remove,omitempty"` // Type of the peer that is to be removed.
Location *GeoLocation `json:"location,omitempty"` // Geographical location of the peer.
Connected *time.Time `json:"connected,omitempty"` // Timestamp of the connection moment.
Disconnected *time.Time `json:"disconnected,omitempty"` // Timestamp of the disonnection moment.
@ -423,9 +456,9 @@ func (db *Dashboard) collectPeerData() {
}
db.peerLock.Unlock()
// diff contains peer events, which trigger operations that
// newPeerEvents contains peer events, which trigger operations that
// will be executed on the peer tree after a metering period.
diff := make([]*PeerEvent, 0, eventLimit)
newPeerEvents := make([]*PeerEvent, 0, eventLimit)
ingress, egress := new(trafficMap), new(trafficMap)
*ingress, *egress = make(trafficMap), make(trafficMap)
@ -436,15 +469,16 @@ func (db *Dashboard) collectPeerData() {
switch event.Type {
case p2p.PeerConnected:
connected := now.Add(-event.Elapsed)
diff = append(diff, &PeerEvent{
newPeerEvents = append(newPeerEvents, &PeerEvent{
IP: event.IP.String(),
ID: event.ID,
ID: event.ID.String(),
Connected: &connected,
})
case p2p.PeerDisconnected:
diff = append(diff, &PeerEvent{
IP: event.IP.String(),
ID: event.ID,
ip, id := event.IP.String(), event.ID.String()
newPeerEvents = append(newPeerEvents, &PeerEvent{
IP: ip,
ID: id,
Disconnected: &now,
})
// The disconnect event comes with the last metered traffic count,
@ -453,11 +487,11 @@ func (db *Dashboard) collectPeerData() {
// period the same peer disconnects multiple times, and appending
// all the samples to the traffic arrays would shift the metering,
// so only the last metering is stored, overwriting the previous one.
ingress.insert(event.IP.String(), event.ID, float64(event.Ingress))
egress.insert(event.IP.String(), event.ID, float64(event.Egress))
ingress.insert(ip, id, float64(event.Ingress))
egress.insert(ip, id, float64(event.Egress))
case p2p.PeerHandshakeFailed:
connected := now.Add(-event.Elapsed)
diff = append(diff, &PeerEvent{
newPeerEvents = append(newPeerEvents, &PeerEvent{
IP: event.IP.String(),
Connected: &connected,
Disconnected: &now,
@ -470,7 +504,9 @@ func (db *Dashboard) collectPeerData() {
p2p.PeerIngressRegistry.Each(collectIngress(ingress))
p2p.PeerEgressRegistry.Each(collectEgress(egress))
// Protect 'peers', because it is part of the history.
db.peerLock.Lock()
// Usually the active peers don't produce events, and marking
// them as active makes it sure that they won't be removed from
// the tree. Only the active peers are registered into the peer
@ -486,44 +522,58 @@ func (db *Dashboard) collectPeerData() {
peers.resetActiveSeparator()
for ip, bundle := range *ingress {
for id := range bundle {
// Only set the peers that are inserted both
// into the ingress and the egress maps.
if _, ok := (*egress)[ip][id]; ok {
peers.setActive(ip, id)
} else {
log.Warn("Peer missing traffic sample", "IP", ip, "ID", id)
}
}
}
var events []*PeerEvent
for i := 0; i < len(diff); i++ {
if diff[i].IP == "" {
log.Warn("Peer event without IP", "event", *diff[i])
var diff []*PeerEvent
for i := 0; i < len(newPeerEvents); i++ {
if newPeerEvents[i].IP == "" {
log.Warn("Peer event without IP", "event", *newPeerEvents[i])
continue
}
diff = append(diff, newPeerEvents[i])
// There are two main branches of peer events coming from the event
// feed, one belongs to the known peers, one to the unknown peers.
// If the event has node ID, it belongs to a known peer, otherwise
// to an unknown one.
if diff[i].ID == "" {
events = append(events, peers.extendUnknown(diff[i])...)
//
// The extension can produce additional peer events, such
// as remove, location and initial samples events.
if newPeerEvents[i].ID == "" {
diff = append(diff, peers.extendUnknown(newPeerEvents[i])...)
continue
}
events = append(events, peers.extendKnown(diff[i])...)
diff = append(diff, peers.extendKnown(newPeerEvents[i])...)
}
// The insertion can produce additional peer events, such
// as remove, location and initial samples events. These
// are stored in the 'events', then appended to the 'diff'.
diff = append(diff, events...)
now := time.Now()
// Update the peer tree using the events.
// Update the peer tree using the traffic maps.
for ip, bundle := range peers.Bundles {
for id, peer := range bundle.KnownPeers {
// Value is 0 if the traffic map doesn't have the
// entry corresponding to the given IP and ID.
curIngress, curEgress := (*ingress)[ip][id], (*egress)[ip][id]
deltaIngress, deltaEgress := curIngress, curEgress
if deltaIngress >= peer.prevIngress {
deltaIngress -= peer.prevIngress
}
if deltaEgress >= peer.prevEgress {
deltaEgress -= peer.prevEgress
}
peer.prevIngress, peer.prevEgress = curIngress, curEgress
i := &ChartEntry{
Time: now,
Value: (*ingress)[ip][id],
Value: deltaIngress,
}
e := &ChartEntry{
Time: now,
Value: (*egress)[ip][id],
Value: deltaEgress,
}
peer.Ingress = append(peer.Ingress, i)
peer.Egress = append(peer.Egress, e)
@ -542,15 +592,20 @@ func (db *Dashboard) collectPeerData() {
})
}
}
//ss, _ := json.MarshalIndent(db.history.Network, "", " ")
//fmt.Println(string(ss))
db.peerLock.Unlock()
//s, _ := json.MarshalIndent(diff, "", " ")
//fmt.Println(string(s))
db.sendToAll(&Message{Network: &NetworkMessage{
Diff: append([]*PeerEvent{}, diff...)},
})
Diff: diff,
}})
// Clear the traffic maps, and the event array,
// prepare them for the next metering.
*ingress, *egress = make(trafficMap), make(trafficMap)
diff = diff[:0]
newPeerEvents = newPeerEvents[:0]
case err := <-subPeer.Err():
log.Warn("Peer subscription error", "err", err)
return

11
p2p/metrics.go
View file

@ -74,7 +74,7 @@ const (
type MeteredPeerEvent struct {
Type MeteredPeerEventType // Type of peer event
IP net.IP // IP address of the peer
ID string // NodeID of the peer
ID enode.ID // NodeID of the peer
Elapsed time.Duration // Time elapsed between the connection and the handshake/disconnection
Ingress uint64 // Ingress count at the moment of the event
Egress uint64 // Egress count at the moment of the event
@ -93,7 +93,7 @@ type meteredConn struct {
connected time.Time // Connection time of the peer
ip net.IP // IP address of the peer
id string // NodeID of the peer
id enode.ID // NodeID of the peer
// trafficMetered denotes if the peer is registered in the traffic registries.
// Its value is true if the metered peer count doesn't reach the limit in the
@ -160,8 +160,7 @@ func (c *meteredConn) Write(b []byte) (n int, err error) {
// handshakeDone is called when a peer handshake is done. Registers the peer to
// the ingress and the egress traffic registries using the peer's IP and node ID,
// also emits connect event.
func (c *meteredConn) handshakeDone(nodeID enode.ID) {
id := nodeID.String()
func (c *meteredConn) handshakeDone(id enode.ID) {
if atomic.AddInt32(&meteredPeerCount, 1) >= MeteredPeerLimit {
// Don't register the peer in the traffic registries.
atomic.AddInt32(&meteredPeerCount, -1)
@ -170,7 +169,7 @@ func (c *meteredConn) handshakeDone(nodeID enode.ID) {
c.lock.Unlock()
log.Warn("Metered peer count reached the limit")
} else {
key := fmt.Sprintf("%s/%s", c.ip, id)
key := fmt.Sprintf("%s/%s", c.ip, id.String())
c.lock.Lock()
c.id, c.trafficMetered = id, true
c.ingressMeter = metrics.NewRegisteredMeter(key, PeerIngressRegistry)
@ -190,7 +189,7 @@ func (c *meteredConn) handshakeDone(nodeID enode.ID) {
func (c *meteredConn) Close() error {
err := c.Conn.Close()
c.lock.RLock()
if c.id == "" {
if c.id == (enode.ID{}) {
// If the peer disconnects before the handshake.
c.lock.RUnlock()
meteredPeerFeed.Send(MeteredPeerEvent{