forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

cmd, dashboard, p2p: send peer info to the dashboard

Browse source
This commit is contained in:
Kurkó Mihály 2019-03-21 19:46:32 +02:00
parent 7dd3194710
commit 12f2efe9d5
13 changed files with 814 additions and 622 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
cmd/geth/config.go
View file

@ -160,7 +160,7 @@ func makeFullNode(ctx *cli.Context) *node.Node {
utils.RegisterEthService(stack, &cfg.Eth)
if ctx.GlobalBool(utils.DashboardEnabledFlag.Name) {
utils.RegisterDashboardService(stack, &cfg.Dashboard, gitCommit)
utils.RegisterDashboardService(stack, &cfg.Dashboard, cfg.Eth.SyncMode, gitCommit)
}
// Whisper must be explicitly enabled by specifying at least 1 whisper flag or in dev mode
shhEnabled := enableWhisper(ctx)

7
cmd/utils/flags.go
View file

@ -1477,9 +1477,12 @@ func RegisterEthService(stack *node.Node, cfg *eth.Config) {
}
// RegisterDashboardService adds a dashboard to the stack.
func RegisterDashboardService(stack *node.Node, cfg *dashboard.Config, commit string) {
func RegisterDashboardService(stack *node.Node, cfg *dashboard.Config, syncMode downloader.SyncMode, commit string) {
stack.Register(func(ctx *node.ServiceContext) (node.Service, error) {
return dashboard.New(cfg, commit, ctx.ResolvePath("logs")), nil
var ethServ *eth.Ethereum
ctx.Service(&ethServ)
return dashboard.New(cfg, syncMode, commit, ctx.ResolvePath("logs")), nil
})
}

648
dashboard/assets.go
View file

File diff suppressed because one or more lines are too long

11
dashboard/assets/components/Dashboard.jsx
View file

@ -83,8 +83,9 @@ const appender = <T>(limit: number, mapper = replacer) => (update: Array<T>, pre
// the execution of unnecessary operations (e.g. copy of the log array).
const defaultContent: () => Content = () => ({
general: {
version: null,
commit: null,
version: null,
commit: null,
syncMode: '',
},
home: {},
chain: {},
@ -119,8 +120,9 @@ const defaultContent: () => Content = () => ({
// TODO (kurkomisi): Define a tricky type which embraces the content and the updaters.
const updaters = {
general: {
version: replacer,
commit: replacer,
version: replacer,
commit: replacer,
syncMode: replacer,
},
home: null,
chain: null,
@ -241,6 +243,7 @@ class Dashboard extends Component<Props, State> {
<div className={this.props.classes.dashboard} style={styles.dashboard}>
<Header
switchSideBar={this.switchSideBar}
syncMode={this.state.content.general.syncMode}
/>
<Body
opened={this.state.sideBar}

3
dashboard/assets/components/Header.jsx
View file

@ -54,8 +54,9 @@ const themeStyles = (theme: Object) => ({
});
export type Props = {
classes: Object, // injected by withStyles()
classes: Object, // injected by withStyles()
switchSideBar: () => void,
syncMode: string,
};
// Header renders the header of the dashboard.

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

@ -36,8 +36,13 @@ import type {Network as NetworkType, PeerEvent} from '../types/content';
import {styles as commonStyles, chartStrokeWidth, hues, hueScale} from '../common';
// Peer chart dimensions.
const trafficChartHeight = 18;
const trafficChartWidth = 400;
const trafficChartHeight = 15;
const trafficChartWidth = 300;
// attemptSeparator separates the peer connection attempts
// such as the peers from the addresses with more attempts
// go to the beginning of the table, and the rest go to the end.
const attemptSeparator = 9;
// setMaxIngress adjusts the peer chart's gradient values based on the given value.
const setMaxIngress = (peer, value) => {
@ -120,6 +125,43 @@ const setEgressChartAttributes = (peer) => {
setMaxEgress(peer, max);
};
// shortName adds some heuristics to the node name in order to make it look meaningful.
const shortName = (name: string) => {
const parts = name.split('/');
if (parts[0].substring(0, 'parity'.length).toLowerCase() === 'parity') {
// Merge Parity and Parity-Ethereum under the same name.
parts[0] = 'Parity';
}
// Cutting anything from the version after the first - or +.
parts[1] = parts[1].split('-')[0].split('+')[0];
return `${parts[0]}/${parts[1]}`;
};
// shortLocation returns a shortened version of the given location object.
const shortLocation = (location: Object) => {
if (!location) {
return '';
}
return `${location.city ? `${location.city}/` : ''}${location.country ? location.country : ''}`;
};
// ethProtocol returns a shortened version of the eth protocol values.
const ethProtocol = (protocols: Object) => {
const {eth} = protocols;
if (typeof eth === 'string') {
return eth;
}
if (!(eth instanceof Object)) {
console.error('Wrong protocol type', eth, typeof eth);
return '';
}
if (!eth.hasOwnProperty('version') || !eth.hasOwnProperty('difficulty') || !eth.hasOwnProperty('head')) {
console.error('Missing protocol attributes', eth);
return '';
}
return `h=${eth.head.substring(0, 10)} v=${eth.version} td=${eth.difficulty}`;
};
// inserter is a state updater function for the main component, which handles the peers.
export const inserter = (sampleLimit: number) => (update: NetworkType, prev: NetworkType) => {
// The first message contains the metered peer history.
@ -134,84 +176,101 @@ export const inserter = (sampleLimit: number) => (update: NetworkType, prev: Net
if (!peer.maxEgress) {
setEgressChartAttributes(peer);
}
if (!peer.name) {
peer.name = '';
peer.shortName = '';
} else if (!peer.shortName) {
peer.shortName = shortName(peer.name);
}
if (!peer.enode) {
peer.enode = '';
}
if (!peer.protocols) {
peer.protocols = {};
}
peer.eth = ethProtocol(peer.protocols);
});
}
bundle.shortLocation = shortLocation(bundle.location);
});
}
if (Array.isArray(update.diff)) {
update.diff.forEach((event: PeerEvent) => {
if (!event.ip) {
console.error('Peer event without IP', event);
if (!event.addr) {
console.error('Peer event without TCP address', event);
return;
}
switch (event.remove) {
case 'bundle': {
delete prev.peers.bundles[event.ip];
delete prev.peers.bundles[event.addr];
return;
}
case 'known': {
if (!event.id) {
console.error('Remove known peer event without ID', event.ip);
if (!event.enode) {
console.error('Remove known peer event without node URL', event.addr);
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);
const bundle = prev.peers.bundles[event.addr];
if (!bundle || !bundle.knownPeers || !bundle.knownPeers[event.enode]) {
console.error('No known peer to remove', event.addr, event.enode);
return;
}
delete bundle.knownPeers[event.id];
return;
}
case 'attempt': {
const bundle = prev.peers.bundles[event.ip];
if (!bundle || !Array.isArray(bundle.attempts) || bundle.attempts.length < 1) {
console.error('No unknown peer to remove', event.ip);
return;
}
bundle.attempts.splice(0, 1);
delete bundle.knownPeers[event.enode];
return;
}
}
if (!prev.peers.bundles[event.ip]) {
prev.peers.bundles[event.ip] = {
if (!prev.peers.bundles[event.addr]) {
prev.peers.bundles[event.addr] = {
location: {
country: '',
city: '',
latitude: 0,
longitude: 0,
},
shortLocation: '',
knownPeers: {},
attempts: [],
attempts: 0,
};
}
const bundle = prev.peers.bundles[event.ip];
const bundle = prev.peers.bundles[event.addr];
if (event.location) {
bundle.location = event.location;
bundle.shortLocation = shortLocation(bundle.location);
return;
}
if (!event.id) {
if (!bundle.attempts) {
bundle.attempts = [];
}
bundle.attempts.push({
connected: event.connected,
disconnected: event.disconnected,
});
if (!event.enode) {
bundle.attempts++;
return;
}
if (!bundle.knownPeers) {
bundle.knownPeers = {};
}
if (!bundle.knownPeers[event.id]) {
bundle.knownPeers[event.id] = {
if (!bundle.knownPeers[event.enode]) {
bundle.knownPeers[event.enode] = {
connected: [],
disconnected: [],
ingress: [],
egress: [],
active: false,
name: '',
shortName: '',
enode: '',
protocols: {},
eth: '',
};
}
const peer = bundle.knownPeers[event.id];
const peer = bundle.knownPeers[event.enode];
if (event.name) {
peer.name = event.name;
peer.shortName = shortName(event.name);
}
if (event.enode) {
peer.enode = event.enode;
}
if (event.protocols) {
peer.protocols = event.protocols;
peer.eth = ethProtocol(peer.protocols);
}
if (!peer.maxIngress) {
setIngressChartAttributes(peer);
}
@ -305,6 +364,11 @@ export const inserter = (sampleLimit: number) => (update: NetworkType, prev: Net
// styles contains the constant styles of the component.
const styles = {
table: {
background: '#212121',
borderCollapse: 'unset',
padding: 5,
},
tableHead: {
height: 'auto',
},
@ -317,9 +381,21 @@ const styles = {
paddingBottom: 0,
paddingLeft: 5,
border: 'none',
fontFamily: 'monospace',
fontSize: 10,
},
};
// limitedWidthStyle returns a style object which cuts the long text with three dots.
const limitedWidthStyle = (width) => {
return {
textOverflow: 'ellipsis',
maxWidth: width,
overflow: 'hidden',
whiteSpace: 'nowrap',
};
};
export type Props = {
container: Object,
content: NetworkType,
@ -351,33 +427,53 @@ class Network extends Component<Props, State> {
return `${month}/${date}/${hours}:${minutes}:${seconds}`;
};
copyToClipboard = (id) => (event) => {
copyToClipboard = (text: string) => (event) => {
event.preventDefault();
navigator.clipboard.writeText(id).then(() => {}, () => {
console.error("Failed to copy node id", id);
navigator.clipboard.writeText(text).then(() => {}, () => {
console.error("Failed to copy", text);
});
};
peerTableRow = (ip, id, bundle, peer) => {
knownPeerTableRow = (addr, enode, bundle, peer) => {
const ingressValues = peer.ingress.map(({value}) => ({ingress: value || 0.001}));
const egressValues = peer.egress.map(({value}) => ({egress: -value || -0.001}));
return (
<TableRow key={`known_${ip}_${id}`} style={styles.tableRow}>
<TableRow key={`known_${addr}_${enode}`} style={styles.tableRow}>
<TableCell style={styles.tableCell}>
{peer.active
? <FontAwesomeIcon icon={fasCircle} color='green' />
: <FontAwesomeIcon icon={farCircle} style={commonStyles.light} />
}
</TableCell>
<TableCell style={{fontFamily: 'monospace', cursor: 'copy', ...styles.tableCell, ...commonStyles.light}} onClick={this.copyToClipboard(id)}>
{id.substring(0, 10)}
<TableCell
style={{
cursor: 'copy',
...styles.tableCell,
...commonStyles.light,
}}
onClick={this.copyToClipboard(enode)}
>
{enode.substring(8, 16)}
</TableCell>
<TableCell style={styles.tableCell}>
{bundle.location ? (() => {
const l = bundle.location;
return `${l.country ? l.country : ''}${l.city ? `/${l.city}` : ''}`;
})() : ''}
<TableCell
style={{
cursor: 'copy',
...limitedWidthStyle(120),
...styles.tableCell,
}}
onClick={this.copyToClipboard(peer.name)}
>
{peer.shortName}
</TableCell>
<TableCell
style={{
cursor: 'copy',
...limitedWidthStyle(120),
...styles.tableCell,
}}
onClick={this.copyToClipboard(JSON.stringify(bundle.location))}
>
{bundle.shortLocation}
</TableCell>
<TableCell style={styles.tableCell}>
<AreaChart
@ -385,14 +481,14 @@ class Network extends Component<Props, State> {
height={trafficChartHeight}
data={ingressValues}
margin={{top: 5, right: 5, bottom: 0, left: 5}}
syncId={`peerIngress_${ip}_${id}`}
syncId={`peerIngress_${addr}_${enode}`}
>
<defs>
<linearGradient id={`ingressGradient_${ip}_${id}`} x1='0' y1='1' x2='0' y2='0'>
<linearGradient id={`ingressGradient_${addr}_${enode}`} x1='0' y1='1' x2='0' y2='0'>
{peer.ingressGradient
&& peer.ingressGradient.map(({offset, color}, i) => (
<stop
key={`ingressStop_${ip}_${id}_${i}`}
key={`ingressStop_${addr}_${enode}_${i}`}
offset={`${offset}%`}
stopColor={color}
/>
@ -405,7 +501,7 @@ class Network extends Component<Props, State> {
dataKey='ingress'
isAnimationActive={false}
type='monotone'
fill={`url(#ingressGradient_${ip}_${id})`}
fill={`url(#ingressGradient_${addr}_${enode})`}
stroke={peer.ingressGradient[peer.ingressGradient.length - 1].color}
strokeWidth={chartStrokeWidth}
/>
@ -415,14 +511,14 @@ class Network extends Component<Props, State> {
height={trafficChartHeight}
data={egressValues}
margin={{top: 0, right: 5, bottom: 5, left: 5}}
syncId={`peerIngress_${ip}_${id}`}
syncId={`peerIngress_${addr}_${enode}`}
>
<defs>
<linearGradient id={`egressGradient_${ip}_${id}`} x1='0' y1='1' x2='0' y2='0'>
<linearGradient id={`egressGradient_${addr}_${enode}`} x1='0' y1='1' x2='0' y2='0'>
{peer.egressGradient
&& peer.egressGradient.map(({offset, color}, i) => (
<stop
key={`egressStop_${ip}_${id}_${i}`}
key={`egressStop_${addr}_${enode}_${i}`}
offset={`${offset}%`}
stopColor={color}
/>
@ -435,91 +531,116 @@ class Network extends Component<Props, State> {
dataKey='egress'
isAnimationActive={false}
type='monotone'
fill={`url(#egressGradient_${ip}_${id})`}
fill={`url(#egressGradient_${addr}_${enode})`}
stroke={peer.egressGradient[0].color}
strokeWidth={chartStrokeWidth}
/>
</AreaChart>
</TableCell>
<TableCell
style={{cursor: 'copy', ...styles.tableCell}}
onClick={this.copyToClipboard(JSON.stringify(peer.protocols.eth))}
>
{peer.eth}
</TableCell>
</TableRow>
);
};
connectionAttemptTableRow = (addr, bundle) => (
<TableRow key={`attempt_${addr}`} style={styles.tableRow}>
<TableCell
style={{cursor: 'copy', ...styles.tableCell}}
onClick={this.copyToClipboard(addr)}
>
{addr}
</TableCell>
<TableCell
style={{cursor: 'copy', ...limitedWidthStyle(120), ...styles.tableCell}}
onClick={this.copyToClipboard(JSON.stringify(bundle.location))}
>
{bundle.shortLocation}
</TableCell>
<TableCell style={styles.tableCell}>
{bundle.attempts}
</TableCell>
</TableRow>
);
render() {
return (
<Grid container direction='row' justify='space-between'>
<Grid item>
<Table>
<Table style={styles.table}>
<TableHead style={styles.tableHead}>
<TableRow style={styles.tableRow}>
<TableCell style={styles.tableCell} />
<TableCell style={styles.tableCell}>Node ID</TableCell>
<TableCell style={styles.tableCell}>Node URL</TableCell>
<TableCell style={styles.tableCell}>Name</TableCell>
<TableCell style={styles.tableCell}>Location</TableCell>
<TableCell style={styles.tableCell}>Traffic</TableCell>
<TableCell style={styles.tableCell}>ETH protocol</TableCell>
</TableRow>
</TableHead>
<TableBody>
{Object.entries(this.props.content.peers.bundles).map(([ip, bundle]) => {
{Object.entries(this.props.content.peers.bundles).map(([addr, bundle]) => {
if (!bundle.knownPeers || Object.keys(bundle.knownPeers).length < 1) {
return null;
}
return Object.entries(bundle.knownPeers).map(([id, peer]) => {
return Object.entries(bundle.knownPeers).map(([enode, peer]) => {
if (peer.active === false) {
return null;
}
return this.peerTableRow(ip, id, bundle, peer);
return this.knownPeerTableRow(addr, enode, bundle, peer);
});
})}
</TableBody>
<TableBody>
{Object.entries(this.props.content.peers.bundles).map(([ip, bundle]) => {
{Object.entries(this.props.content.peers.bundles).map(([addr, bundle]) => {
if (!bundle.knownPeers || Object.keys(bundle.knownPeers).length < 1) {
return null;
}
return Object.entries(bundle.knownPeers).map(([id, peer]) => {
return Object.entries(bundle.knownPeers).map(([enode, peer]) => {
if (peer.active === true) {
return null;
}
return this.peerTableRow(ip, id, bundle, peer);
return this.knownPeerTableRow(addr, enode, bundle, peer);
});
})}
</TableBody>
</Table>
</Grid>
<Grid item>
<Typography variant='subtitle1' gutterBottom>
Connection attempts
</Typography>
<Table>
<TableHead style={styles.tableHead}>
<TableRow style={styles.tableRow}>
<TableCell style={styles.tableCell}>IP</TableCell>
<TableCell style={styles.tableCell}>Location</TableCell>
<TableCell style={styles.tableCell}>Nr</TableCell>
</TableRow>
</TableHead>
<TableBody>
{Object.entries(this.props.content.peers.bundles).map(([ip, bundle]) => {
if (!bundle.attempts || bundle.attempts.length < 1) {
return null;
}
return (
<TableRow key={`attempt_${ip}`} style={styles.tableRow}>
<TableCell style={styles.tableCell}>{ip}</TableCell>
<TableCell style={styles.tableCell}>
{bundle.location ? (() => {
const l = bundle.location;
return `${l.country ? l.country : ''}${l.city ? `/${l.city}` : ''}`;
})() : ''}
</TableCell>
<TableCell style={styles.tableCell}>
{Object.values(bundle.attempts).length}
</TableCell>
</TableRow>
);
})}
</TableBody>
</Table>
<div style={styles.table}>
<Typography variant='subtitle1' gutterBottom>
Connection attempts
</Typography>
<Table>
<TableHead style={styles.tableHead}>
<TableRow style={styles.tableRow}>
<TableCell style={styles.tableCell}>TCP address</TableCell>
<TableCell style={styles.tableCell}>Location</TableCell>
<TableCell style={styles.tableCell}>Nr</TableCell>
</TableRow>
</TableHead>
<TableBody>
{Object.entries(this.props.content.peers.bundles).map(([addr, bundle]) => {
if (!bundle.attempts || bundle.attempts <= attemptSeparator) {
return null;
}
return this.connectionAttemptTableRow(addr, bundle);
})}
</TableBody>
<TableBody>
{Object.entries(this.props.content.peers.bundles).map(([addr, bundle]) => {
if (!bundle.attempts || bundle.attempts < 1 || bundle.attempts > attemptSeparator) {
return null;
}
return this.connectionAttemptTableRow(addr, bundle);
})}
</TableBody>
</Table>
</div>
</Grid>
</Grid>
);

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

@ -33,8 +33,9 @@ export type ChartEntry = {
};
export type General = {
version: ?string,
commit: ?string,
version: ?string,
commit: ?string,
syncMode: string,
};
export type Home = {
@ -55,8 +56,10 @@ export type Network = {
};
export type PeerEvent = {
ip: string,
id: string,
name: string,
addr: string,
enode: string,
protocols: {[string]: Object},
remove: string,
location: GeoLocation,
connected: Date,
@ -71,9 +74,9 @@ export type Peers = {
};
export type PeerBundle = {
location: GeoLocation,
knownPeers: {[string]: KnownPeer},
attempts: Array<UnknownPeer>,
location: GeoLocation,
knownPeers: {[string]: KnownPeer},
attempts: number,
};
export type KnownPeer = {
@ -81,14 +84,12 @@ export type KnownPeer = {
disconnected: Array<Date>,
ingress: Array<ChartEntries>,
egress: Array<ChartEntries>,
name: string,
enode: string,
protocols: {[string]: Object},
active: boolean,
};
export type UnknownPeer = {
connected: Date,
disconnected: Date,
};
export type GeoLocation = {
country: string,
city: string,

47
dashboard/dashboard.go
View file

@ -27,13 +27,14 @@ package dashboard
import (
"fmt"
"io"
"net"
"net/http"
"sync"
"sync/atomic"
"time"
"io"
"github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
@ -67,6 +68,11 @@ type Dashboard struct {
quit chan chan error // Channel used for graceful exit
wg sync.WaitGroup // Wait group used to close the data collector threads
peerCh chan p2p.MeteredPeerEvent // Peer event channel.
subPeer event.Subscription // Peer event subscription.
syncMode downloader.SyncMode
}
// client represents active websocket connection with a remote browser.
@ -77,8 +83,13 @@ type client struct {
}
// New creates a new dashboard instance with the given configuration.
func New(config *Config, commit string, logdir string) *Dashboard {
now := time.Now()
func New(config *Config, syncMode downloader.SyncMode, commit string, logdir string) *Dashboard {
// There is a data race between the network layer and the dashboard, which
// can cause some lost peer events, therefore some peers might not appear
// on the dashboard.
// In order to solve this problem, the peer event subscription is registered
// here, before the network layer starts.
peerCh := make(chan p2p.MeteredPeerEvent, p2p.MeteredPeerLimit)
versionMeta := ""
if len(params.VersionMeta) > 0 {
versionMeta = fmt.Sprintf(" (%s)", params.VersionMeta)
@ -89,26 +100,30 @@ func New(config *Config, commit string, logdir string) *Dashboard {
quit: make(chan chan error),
history: &Message{
General: &GeneralMessage{
Commit: commit,
Version: fmt.Sprintf("v%d.%d.%d%s", params.VersionMajor, params.VersionMinor, params.VersionPatch, versionMeta),
Commit: commit,
Version: fmt.Sprintf("v%d.%d.%d%s", params.VersionMajor, params.VersionMinor, params.VersionPatch, versionMeta),
SyncMode: syncMode.String(),
},
System: &SystemMessage{
ActiveMemory: emptyChartEntries(now, sampleLimit),
VirtualMemory: emptyChartEntries(now, sampleLimit),
NetworkIngress: emptyChartEntries(now, sampleLimit),
NetworkEgress: emptyChartEntries(now, sampleLimit),
ProcessCPU: emptyChartEntries(now, sampleLimit),
SystemCPU: emptyChartEntries(now, sampleLimit),
DiskRead: emptyChartEntries(now, sampleLimit),
DiskWrite: emptyChartEntries(now, sampleLimit),
ActiveMemory: emptyChartEntries(sampleLimit),
VirtualMemory: emptyChartEntries(sampleLimit),
NetworkIngress: emptyChartEntries(sampleLimit),
NetworkEgress: emptyChartEntries(sampleLimit),
ProcessCPU: emptyChartEntries(sampleLimit),
SystemCPU: emptyChartEntries(sampleLimit),
DiskRead: emptyChartEntries(sampleLimit),
DiskWrite: emptyChartEntries(sampleLimit),
},
},
logdir: logdir,
logdir: logdir,
peerCh: peerCh,
subPeer: p2p.SubscribeMeteredPeerEvent(peerCh),
syncMode: syncMode,
}
}
// emptyChartEntries returns a ChartEntry array containing limit number of empty samples.
func emptyChartEntries(t time.Time, limit int) ChartEntries {
func emptyChartEntries(limit int) ChartEntries {
ce := make(ChartEntries, limit)
for i := 0; i < limit; i++ {
ce[i] = new(ChartEntry)

5
dashboard/message.go
View file

@ -37,8 +37,9 @@ type ChartEntry struct {
}
type GeneralMessage struct {
Version string `json:"version,omitempty"`
Commit string `json:"commit,omitempty"`
Version string `json:"version,omitempty"`
Commit string `json:"commit,omitempty"`
SyncMode string `json:"syncMode,omitempty"`
}
type HomeMessage struct {

276
dashboard/peers.go
View file

@ -18,6 +18,7 @@ package dashboard
import (
"container/list"
"reflect"
"strings"
"time"
@ -28,9 +29,7 @@ import (
)
const (
eventBufferLimit = 128 // Maximum number of buffered peer events.
knownPeerLimit = 100 // Maximum number of stored peers, which successfully made the handshake.
attemptLimit = 200 // Maximum number of stored peers, which failed to make the handshake.
knownPeerLimit = 100 // Maximum number of stored peers, which successfully made 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
@ -83,14 +82,6 @@ type peerContainer struct {
// inactivePeers contains the peers with closed connection in chronological order.
inactivePeers *list.List
// attemptOrder is the super array containing the IP addresses, from which
// the peers attempted to connect then failed before/during the handshake.
// Its values are appended in chronological order, which means that the
// oldest attempt is at the beginning of the array. When the first element
// is removed, the first element of the related bundle's attempt array is
// removed too, ensuring that always the latest attempts are stored.
attemptOrder []string
// geodb is the geoip database used to retrieve the peers' geographical location.
geodb *geoDB
}
@ -100,7 +91,6 @@ func newPeerContainer(geodb *geoDB) *peerContainer {
return &peerContainer{
Bundles: make(map[string]*peerBundle),
inactivePeers: list.New(),
attemptOrder: make([]string, 0, attemptLimit),
geodb: geodb,
}
}
@ -110,48 +100,62 @@ func newPeerContainer(geodb *geoDB) *peerContainer {
// the IP address from the database and creates a corresponding peer event.
// Returns the bundle belonging to the given IP and the events occurring during
// the initialization.
func (pc *peerContainer) bundle(ip string) (*peerBundle, []*peerEvent) {
func (pc *peerContainer) bundle(addr string) (*peerBundle, []*peerEvent) {
var events []*peerEvent
if _, ok := pc.Bundles[ip]; !ok {
location := pc.geodb.location(ip)
if _, ok := pc.Bundles[addr]; !ok {
i := strings.IndexByte(addr, ':')
if i < 0 {
i = len(addr)
}
location := pc.geodb.location(addr[:i])
events = append(events, &peerEvent{
IP: ip,
Addr: addr,
Location: location,
})
pc.Bundles[ip] = &peerBundle{
pc.Bundles[addr] = &peerBundle{
Location: location,
KnownPeers: make(map[string]*knownPeer),
}
}
return pc.Bundles[ip], events
return pc.Bundles[addr], events
}
// extendKnown handles the events of the successfully connected peers.
// Returns the events occurring during the extension.
func (pc *peerContainer) extendKnown(event *peerEvent) []*peerEvent {
bundle, events := pc.bundle(event.IP)
peer, peerEvents := bundle.knownPeer(event.IP, event.ID)
bundle, events := pc.bundle(event.Addr)
peer, peerEvents := bundle.knownPeer(event.Addr, event.Enode)
events = append(events, peerEvents...)
// Append the connect and the disconnect events to
// the corresponding arrays keeping the limit.
switch {
case event.Connected != nil:
case event.Connected != nil: // Handshake succeeded
peer.Connected = append(peer.Connected, event.Connected)
if first := len(peer.Connected) - sampleLimit; first > 0 {
peer.Connected = peer.Connected[first:]
}
if event.peer == nil {
log.Warn("Peer handshake succeeded event without peer instance", "addr", event.Addr, "enode", event.Enode)
}
peer.peer = event.peer
info := event.peer.Info()
peer.Name = info.Name
peer.Protocols = info.Protocols
peer.Active = true
events = append(events, &peerEvent{
Activity: Active,
IP: peer.ip,
ID: peer.id,
})
e := &peerEvent{
Activity: Active,
Name: info.Name,
Addr: peer.addr,
Enode: peer.enode,
Protocols: peer.Protocols,
}
events = append(events, e)
pc.activeCount++
if peer.listElement != nil {
_ = pc.inactivePeers.Remove(peer.listElement)
peer.listElement = nil
}
case event.Disconnected != nil:
case event.Disconnected != nil: // Peer disconnected
peer.Disconnected = append(peer.Disconnected, event.Disconnected)
if first := len(peer.Disconnected) - sampleLimit; first > 0 {
peer.Disconnected = peer.Disconnected[first:]
@ -159,8 +163,8 @@ func (pc *peerContainer) extendKnown(event *peerEvent) []*peerEvent {
peer.Active = false
events = append(events, &peerEvent{
Activity: Inactive,
IP: peer.ip,
ID: peer.id,
Addr: peer.addr,
Enode: peer.enode,
})
pc.activeCount--
if peer.listElement != nil {
@ -169,12 +173,14 @@ func (pc *peerContainer) extendKnown(event *peerEvent) []*peerEvent {
}
// Insert the peer into the list.
peer.listElement = pc.inactivePeers.PushBack(peer)
default:
log.Warn("Unexpected known peer event", "event", *event)
}
for pc.inactivePeers.Len() > 0 && pc.activeCount+pc.inactivePeers.Len() > knownPeerLimit {
// While the count of the known peers is greater than the limit,
// remove the first element from the inactive peer list and from the map.
if removedPeer, ok := pc.inactivePeers.Remove(pc.inactivePeers.Front()).(*knownPeer); ok {
events = append(events, pc.removeKnown(removedPeer.ip, removedPeer.id)...)
events = append(events, pc.removeKnown(removedPeer.addr, removedPeer.enode)...)
} else {
log.Warn("Failed to parse the removed peer")
}
@ -185,25 +191,6 @@ func (pc *peerContainer) extendKnown(event *peerEvent) []*peerEvent {
return events
}
// handleAttempt handles the events of the peers failing before/during the handshake.
// Returns the events occurring during the extension.
func (pc *peerContainer) handleAttempt(event *peerEvent) []*peerEvent {
bundle, events := pc.bundle(event.IP)
bundle.Attempts = append(bundle.Attempts, &peerAttempt{
Connected: *event.Connected,
Disconnected: *event.Disconnected,
})
pc.attemptOrder = append(pc.attemptOrder, event.IP)
for len(pc.attemptOrder) > attemptLimit {
// 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.
events = append(events, pc.removeAttempt(pc.attemptOrder[0])...)
pc.attemptOrder = pc.attemptOrder[1:]
}
return events
}
// peerBundle contains the peers belonging to a given IP address.
type peerBundle struct {
// Location contains the geographical location based on the bundle's IP address.
@ -213,57 +200,35 @@ type peerBundle struct {
// maintainer data structure using the node ID as key.
KnownPeers map[string]*knownPeer `json:"knownPeers,omitempty"`
// Attempts contains the failed connection attempts of the
// peers belonging to a given IP address in chronological order.
Attempts []*peerAttempt `json:"attempts,omitempty"`
// Attempts contains the count of the failed connection
// attempts of the peers belonging to a given IP address.
Attempts uint `json:"attempts,omitempty"`
}
// 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) (events []*peerEvent) {
func (pc *peerContainer) removeKnown(addr, enode string) (events []*peerEvent) {
// TODO (kurkomisi): Remove peers that don't have traffic samples anymore.
if bundle, ok := pc.Bundles[ip]; ok {
if _, ok := bundle.KnownPeers[id]; ok {
if bundle, ok := pc.Bundles[addr]; ok {
if _, ok := bundle.KnownPeers[enode]; ok {
events = append(events, &peerEvent{
Remove: RemoveKnown,
IP: ip,
ID: id,
Addr: addr,
Enode: enode,
})
delete(bundle.KnownPeers, id)
delete(bundle.KnownPeers, enode)
} else {
log.Warn("No peer to remove", "ip", ip, "id", id)
log.Warn("No peer to remove", "addr", addr, "enode", enode)
}
if len(bundle.KnownPeers) < 1 && len(bundle.Attempts) < 1 {
if len(bundle.KnownPeers) < 1 && bundle.Attempts < 1 {
events = append(events, &peerEvent{
Remove: RemoveBundle,
IP: ip,
Addr: addr,
})
delete(pc.Bundles, ip)
delete(pc.Bundles, addr)
}
} else {
log.Warn("No bundle to remove", "ip", ip)
}
return events
}
// removeAttempt removes the peer attempt belonging to the
// given IP address and node ID from the peer tree.
func (pc *peerContainer) removeAttempt(ip string) (events []*peerEvent) {
if bundle, ok := pc.Bundles[ip]; ok {
if len(bundle.Attempts) > 0 {
events = append(events, &peerEvent{
Remove: RemoveAttempt,
IP: ip,
})
bundle.Attempts = bundle.Attempts[1:]
}
if len(bundle.Attempts) < 1 && len(bundle.KnownPeers) < 1 {
events = append(events, &peerEvent{
Remove: RemoveBundle,
IP: ip,
})
delete(pc.Bundles, ip)
}
log.Warn("No bundle to remove", "addr", addr)
}
return events
}
@ -272,26 +237,25 @@ func (pc *peerContainer) removeAttempt(ip string) (events []*peerEvent) {
// to the given IP address and node ID wasn't metered so far. Returns the peer
// belonging to the given IP and ID as well as the events occurring during the
// initialization.
func (bundle *peerBundle) knownPeer(ip, id string) (*knownPeer, []*peerEvent) {
func (bundle *peerBundle) knownPeer(addr, enode string) (*knownPeer, []*peerEvent) {
var events []*peerEvent
if _, ok := bundle.KnownPeers[id]; !ok {
now := time.Now()
ingress := emptyChartEntries(now, sampleLimit)
egress := emptyChartEntries(now, sampleLimit)
if _, ok := bundle.KnownPeers[enode]; !ok {
ingress := emptyChartEntries(sampleLimit)
egress := emptyChartEntries(sampleLimit)
events = append(events, &peerEvent{
IP: ip,
ID: id,
Addr: addr,
Enode: enode,
Ingress: append([]*ChartEntry{}, ingress...),
Egress: append([]*ChartEntry{}, egress...),
})
bundle.KnownPeers[id] = &knownPeer{
ip: ip,
id: id,
bundle.KnownPeers[enode] = &knownPeer{
addr: addr,
enode: enode,
Ingress: ingress,
Egress: egress,
}
}
return bundle.KnownPeers[id], events
return bundle.KnownPeers[enode], events
}
// knownPeer contains the metered data of a particular peer.
@ -312,31 +276,26 @@ type knownPeer struct {
Ingress ChartEntries `json:"ingress,omitempty"`
Egress ChartEntries `json:"egress,omitempty"`
Name string `json:"name,omitempty"` // Name of the node, including client type, version, OS, custom data
Enode string `json:"enode,omitempty"` // Node URL
Protocols map[string]interface{} `json:"protocols,omitempty"` // Sub-protocol specific metadata fields
Active bool `json:"active"` // Denotes if the peer is still connected.
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.
addr, enode string // The IP and the ID by which the peer can be accessed in the tree.
prevIngress float64
prevEgress float64
}
// peerAttempt contains a failed peer connection attempt's attributes.
type peerAttempt struct {
// Connected contains the timestamp of the connection attempt's moment.
Connected time.Time `json:"connected"`
// Disconnected contains the timestamp of the
// moment when the connection attempt failed.
Disconnected time.Time `json:"disconnected"`
peer *p2p.Peer // Connected remote node instance
}
type RemovedPeerType string
type ActivityType string
const (
RemoveKnown RemovedPeerType = "known"
RemoveAttempt RemovedPeerType = "attempt"
RemoveBundle RemovedPeerType = "bundle"
RemoveKnown RemovedPeerType = "known"
RemoveBundle RemovedPeerType = "bundle"
Active ActivityType = "active"
Inactive ActivityType = "inactive"
@ -344,15 +303,19 @@ const (
// 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.
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.
Ingress ChartEntries `json:"ingress,omitempty"` // Ingress samples.
Egress ChartEntries `json:"egress,omitempty"` // Egress samples.
Activity ActivityType `json:"activity,omitempty"` // Connection status change.
Name string `json:"name,omitempty"` // Name of the node, including client type, version, OS, custom data
Addr string `json:"addr,omitempty"` // TCP address of the peer.
Enode string `json:"enode,omitempty"` // Node URL
Protocols map[string]interface{} `json:"protocols,omitempty"` // Sub-protocol specific metadata fields
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.
Ingress ChartEntries `json:"ingress,omitempty"` // Ingress samples.
Egress ChartEntries `json:"egress,omitempty"` // Egress samples.
Activity ActivityType `json:"activity,omitempty"` // Connection status change.
peer *p2p.Peer // Connected remote node instance.
}
// trafficMap is a container for the periodically collected peer traffic.
@ -380,10 +343,6 @@ func (db *Dashboard) collectPeerData() {
}
defer db.geodb.close()
peerCh := make(chan p2p.MeteredPeerEvent, eventBufferLimit) // Peer event channel.
subPeer := p2p.SubscribeMeteredPeerEvent(peerCh) // Subscribe to peer events.
defer subPeer.Unsubscribe() // Unsubscribe at the end.
ticker := time.NewTicker(db.config.Refresh)
defer ticker.Stop()
@ -400,11 +359,11 @@ func (db *Dashboard) collectPeerData() {
// The function which can be passed to the registry.
return func(name string, i interface{}) {
if m, ok := i.(metrics.Meter); ok {
// The name of the meter has the format: <common traffic prefix><IP>/<ID>
if k := strings.Split(strings.TrimPrefix(name, prefix), "/"); len(k) == 2 {
traffic.insert(k[0], k[1], float64(m.Count()))
enode := strings.TrimPrefix(name, prefix)
if addr := strings.Split(enode, "@"); len(addr) == 2 {
traffic.insert(addr[1], enode, float64(m.Count()))
} else {
log.Warn("Invalid meter name", "name", name, "prefix", prefix)
log.Warn("Invalid enode", "enode", enode)
}
} else {
log.Warn("Invalid meter type", "name", name)
@ -428,23 +387,32 @@ func (db *Dashboard) collectPeerData() {
ingress, egress := new(trafficMap), new(trafficMap)
*ingress, *egress = make(trafficMap), make(trafficMap)
defer db.subPeer.Unsubscribe()
for {
select {
case event := <-peerCh:
case event := <-db.peerCh:
now := time.Now()
switch event.Type {
case p2p.PeerConnected:
case p2p.PeerHandshakeFailed:
connected := now.Add(-event.Elapsed)
newPeerEvents = append(newPeerEvents, &peerEvent{
IP: event.IP.String(),
ID: event.ID.String(),
Addr: event.Addr,
Connected: &connected,
Disconnected: &now,
})
case p2p.PeerHandshakeSucceeded:
connected := now.Add(-event.Elapsed)
newPeerEvents = append(newPeerEvents, &peerEvent{
Addr: event.Addr,
Enode: event.Enode,
peer: event.Peer,
Connected: &connected,
})
case p2p.PeerDisconnected:
ip, id := event.IP.String(), event.ID.String()
addr, enode := event.Addr, event.Enode
newPeerEvents = append(newPeerEvents, &peerEvent{
IP: ip,
ID: id,
Addr: addr,
Enode: enode,
Disconnected: &now,
})
// The disconnect event comes with the last metered traffic count,
@ -453,15 +421,8 @@ 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(ip, id, float64(event.Ingress))
egress.insert(ip, id, float64(event.Egress))
case p2p.PeerHandshakeFailed:
connected := now.Add(-event.Elapsed)
newPeerEvents = append(newPeerEvents, &peerEvent{
IP: event.IP.String(),
Connected: &connected,
Disconnected: &now,
})
ingress.insert(addr, enode, float64(event.Ingress))
egress.insert(addr, enode, float64(event.Egress))
default:
log.Error("Unknown metered peer event type", "type", event.Type)
}
@ -475,7 +436,7 @@ func (db *Dashboard) collectPeerData() {
var diff []*peerEvent
for i := 0; i < len(newPeerEvents); i++ {
if newPeerEvents[i].IP == "" {
if newPeerEvents[i].Addr == "" {
log.Warn("Peer event without IP", "event", *newPeerEvents[i])
continue
}
@ -487,18 +448,20 @@ func (db *Dashboard) collectPeerData() {
//
// The extension can produce additional peer events, such
// as remove, location and initial samples events.
if newPeerEvents[i].ID == "" {
diff = append(diff, peers.handleAttempt(newPeerEvents[i])...)
if newPeerEvents[i].Enode == "" {
bundle, events := peers.bundle(newPeerEvents[i].Addr)
bundle.Attempts++
diff = append(diff, events...)
continue
}
diff = append(diff, peers.extendKnown(newPeerEvents[i])...)
}
// Update the peer tree using the traffic maps.
for ip, bundle := range peers.Bundles {
for id, peer := range bundle.KnownPeers {
for addr, bundle := range peers.Bundles {
for enode, 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]
curIngress, curEgress := (*ingress)[addr][enode], (*egress)[addr][enode]
deltaIngress, deltaEgress := curIngress, curEgress
if deltaIngress >= peer.prevIngress {
deltaIngress -= peer.prevIngress
@ -523,11 +486,22 @@ func (db *Dashboard) collectPeerData() {
}
// Creating the traffic sample events.
diff = append(diff, &peerEvent{
IP: ip,
ID: id,
Addr: addr,
Enode: enode,
Ingress: ChartEntries{i},
Egress: ChartEntries{e},
})
if peer.peer != nil {
info := peer.peer.Info()
if !reflect.DeepEqual(peer.Protocols, info.Protocols) {
peer.Protocols = info.Protocols
diff = append(diff, &peerEvent{
Addr: addr,
Enode: enode,
Protocols: peer.Protocols,
})
}
}
}
}
db.peerLock.Unlock()
@ -541,7 +515,7 @@ func (db *Dashboard) collectPeerData() {
// prepare them for the next metering.
*ingress, *egress = make(trafficMap), make(trafficMap)
newPeerEvents = newPeerEvents[:0]
case err := <-subPeer.Err():
case err := <-db.subPeer.Err():
log.Warn("Peer subscription error", "err", err)
return
case errc := <-db.quit:

2
p2p/dial.go
View file

@ -350,7 +350,7 @@ func (t *dialTask) dial(srv *Server, dest *enode.Node) error {
if err != nil {
return &dialError{err}
}
mfd := newMeteredConn(fd, false, dest.IP())
mfd := newMeteredConn(fd, false, &net.TCPAddr{IP: dest.IP(), Port: dest.TCP()})
return srv.SetupConn(mfd, t.flags, dest)
}

83
p2p/metrics.go
View file

@ -19,14 +19,11 @@
package p2p
import (
"fmt"
"net"
"sync"
"sync/atomic"
"time"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
@ -58,24 +55,25 @@ var (
type MeteredPeerEventType int
const (
// PeerConnected is the type of event emitted when a peer successfully
// made the handshake.
PeerConnected MeteredPeerEventType = iota
// PeerHandshakeSucceeded is the type of event
// emitted when a peer successfully makes the handshake.
PeerHandshakeSucceeded MeteredPeerEventType = iota
// PeerHandshakeFailed is the type of event emitted when a peer fails to
// make the encryption handshake or disconnects before it.
PeerHandshakeFailed
// PeerDisconnected is the type of event emitted when a peer disconnects.
PeerDisconnected
// PeerHandshakeFailed is the type of event emitted when a peer fails to
// make the handshake or disconnects before the handshake.
PeerHandshakeFailed
)
// MeteredPeerEvent is an event emitted when peers connect or disconnect.
type MeteredPeerEvent struct {
Type MeteredPeerEventType // Type of peer event
IP net.IP // IP address of the peer
ID enode.ID // NodeID of the peer
Addr string // TCP address of the peer
Elapsed time.Duration // Time elapsed between the connection and the handshake/disconnection
Enode string // Node URL
Peer *Peer // Connected remote node instance
Ingress uint64 // Ingress count at the moment of the event
Egress uint64 // Egress count at the moment of the event
}
@ -91,9 +89,9 @@ func SubscribeMeteredPeerEvent(ch chan<- MeteredPeerEvent) event.Subscription {
type meteredConn struct {
net.Conn // Network connection to wrap with metering
connected time.Time // Connection time of the peer
ip net.IP // IP address of the peer
id enode.ID // NodeID of the peer
connected time.Time // Connection time of the peer
addr *net.TCPAddr // TCP address of the peer
enode string // Node URL 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
@ -109,13 +107,13 @@ type meteredConn struct {
// connection meter and also increases the metered peer count. If the metrics
// system is disabled or the IP address is unspecified, this function returns
// the original object.
func newMeteredConn(conn net.Conn, ingress bool, ip net.IP) net.Conn {
func newMeteredConn(conn net.Conn, ingress bool, addr *net.TCPAddr) net.Conn {
// Short circuit if metrics are disabled
if !metrics.Enabled {
return conn
}
if ip.IsUnspecified() {
log.Warn("Peer IP is unspecified")
if addr == nil || addr.IP.IsUnspecified() {
log.Warn("Peer address is unspecified")
return conn
}
// Bump the connection counters and wrap the connection
@ -126,7 +124,7 @@ func newMeteredConn(conn net.Conn, ingress bool, ip net.IP) net.Conn {
}
return &meteredConn{
Conn: conn,
ip: ip,
addr: addr,
connected: time.Now(),
}
}
@ -157,30 +155,28 @@ func (c *meteredConn) Write(b []byte) (n int, err error) {
return n, err
}
// 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(id enode.ID) {
// TODO (kurkomisi): use the node URL instead of the pure node ID. (the String() method of *Node)
// handshakeDone is called after the connection passes the handshake.
func (c *meteredConn) handshakeDone(peer *Peer) {
enode := peer.Node().String()
if atomic.AddInt32(&meteredPeerCount, 1) >= MeteredPeerLimit {
// Don't register the peer in the traffic registries.
atomic.AddInt32(&meteredPeerCount, -1)
c.lock.Lock()
c.id, c.trafficMetered = id, false
c.enode, c.trafficMetered = enode, false
c.lock.Unlock()
log.Warn("Metered peer count reached the limit")
} else {
key := fmt.Sprintf("%s/%s", c.ip, id.String())
c.lock.Lock()
c.id, c.trafficMetered = id, true
c.ingressMeter = metrics.NewRegisteredMeter(key, PeerIngressRegistry)
c.egressMeter = metrics.NewRegisteredMeter(key, PeerEgressRegistry)
c.enode, c.trafficMetered = enode, true
c.ingressMeter = metrics.NewRegisteredMeter(enode, PeerIngressRegistry)
c.egressMeter = metrics.NewRegisteredMeter(enode, PeerEgressRegistry)
c.lock.Unlock()
}
meteredPeerFeed.Send(MeteredPeerEvent{
Type: PeerConnected,
IP: c.ip,
ID: id,
Type: PeerHandshakeSucceeded,
Addr: c.addr.String(),
Enode: enode,
Peer: peer,
Elapsed: time.Since(c.connected),
})
}
@ -190,24 +186,24 @@ func (c *meteredConn) handshakeDone(id enode.ID) {
func (c *meteredConn) Close() error {
err := c.Conn.Close()
c.lock.RLock()
if c.id == (enode.ID{}) {
// If the peer disconnects before the handshake.
if c.enode == "" {
// If the peer disconnects before/during the handshake.
c.lock.RUnlock()
meteredPeerFeed.Send(MeteredPeerEvent{
Type: PeerHandshakeFailed,
IP: c.ip,
Addr: c.addr.String(),
Elapsed: time.Since(c.connected),
})
return err
}
id := c.id
enode := c.enode
if !c.trafficMetered {
// If the peer isn't registered in the traffic registries.
c.lock.RUnlock()
meteredPeerFeed.Send(MeteredPeerEvent{
Type: PeerDisconnected,
IP: c.ip,
ID: id,
Type: PeerDisconnected,
Addr: c.addr.String(),
Enode: enode,
})
return err
}
@ -218,14 +214,13 @@ func (c *meteredConn) Close() error {
atomic.AddInt32(&meteredPeerCount, -1)
// Unregister the peer from the traffic registries
key := fmt.Sprintf("%s/%s", c.ip, id)
PeerIngressRegistry.Unregister(key)
PeerEgressRegistry.Unregister(key)
PeerIngressRegistry.Unregister(enode)
PeerEgressRegistry.Unregister(enode)
meteredPeerFeed.Send(MeteredPeerEvent{
Type: PeerDisconnected,
IP: c.ip,
ID: id,
Addr: c.addr.String(),
Enode: enode,
Ingress: ingress,
Egress: egress,
})

12
p2p/server.go
View file

@ -736,6 +736,9 @@ running:
if p.Inbound() {
inboundCount++
}
if conn, ok := c.fd.(*meteredConn); ok {
conn.handshakeDone(p)
}
}
// The dialer logic relies on the assumption that
// dial tasks complete after the peer has been added or
@ -863,11 +866,11 @@ func (srv *Server) listenLoop() {
}
}
var ip net.IP
var addr *net.TCPAddr
if tcp, ok := fd.RemoteAddr().(*net.TCPAddr); ok {
ip = tcp.IP
addr = tcp
}
fd = newMeteredConn(fd, true, ip)
fd = newMeteredConn(fd, true, addr)
srv.log.Trace("Accepted connection", "addr", fd.RemoteAddr())
go func() {
srv.SetupConn(fd, inboundConn, nil)
@ -920,9 +923,6 @@ func (srv *Server) setupConn(c *conn, flags connFlag, dialDest *enode.Node) erro
} else {
c.node = nodeFromConn(remotePubkey, c.fd)
}
if conn, ok := c.fd.(*meteredConn); ok {
conn.handshakeDone(c.node.ID())
}
clog := srv.log.New("id", c.node.ID(), "addr", c.fd.RemoteAddr(), "conn", c.flags)
err = srv.checkpoint(c, srv.posthandshake)
if err != nil {