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

dashboard, p2p: separate active peers from inactive ones

Browse source
This commit is contained in:
Kurkó Mihály 2018-11-14 22:39:52 +02:00
parent 92d5937023
commit 547bad0d60
12 changed files with 13589 additions and 13946 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24609
dashboard/assets.go
View file

File diff suppressed because one or more lines are too long

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

@ -23,6 +23,7 @@ import TableHead from '@material-ui/core/TableHead';
import TableBody from '@material-ui/core/TableBody';
import TableRow from '@material-ui/core/TableRow';
import TableCell from '@material-ui/core/TableCell';
import Grid from '@material-ui/core/Grid/Grid';
import AreaChart from 'recharts/es6/chart/AreaChart';
import Tooltip from 'recharts/es6/component/Tooltip';
import Area from 'recharts/es6/cartesian/Area';
@ -57,13 +58,13 @@ export const inserter = (sampleLimit: number) => (update: NetworkType, prev: Net
delete bundle.knownPeers[event.id];
return;
}
case 'unknown': {
case 'attempt': {
const bundle = prev.peers.bundles[event.ip];
if (!bundle || !Array.isArray(bundle.unknownPeers) || bundle.unknownPeers.length < 1) {
if (!bundle || !Array.isArray(bundle.attempts) || bundle.attempts.length < 1) {
console.error('No unknown peer to remove', event.ip);
return;
}
bundle.unknownPeers.splice(0, 1);
bundle.attempts.splice(0, 1);
return;
}
}
@ -76,7 +77,7 @@ export const inserter = (sampleLimit: number) => (update: NetworkType, prev: Net
longitude: 0,
},
knownPeers: {},
unknownPeers: [],
attempts: [],
};
}
const bundle = prev.peers.bundles[event.ip];
@ -85,13 +86,13 @@ export const inserter = (sampleLimit: number) => (update: NetworkType, prev: Net
return;
}
if (!event.id) {
bundle.unknownPeers.push({
bundle.attempts.push({
connected: event.connected,
disconnected: event.disconnected,
});
return;
}
if (!bundle.knownPeers[event.id]) {
if (!bundle.knownPeers || !bundle.knownPeers[event.id]) {
bundle.knownPeers[event.id] = {
connected: [],
disconnected: [],
@ -138,7 +139,7 @@ type State = {};
// Network renders the network page.
class Network extends Component<Props, State> {
formatTime = (t) => {
formatTime = (t: string) => {
const time = new Date(t);
if (isNaN(time)) {
return '';
@ -153,16 +154,14 @@ class Network extends Component<Props, State> {
render() {
return (
<div>
<Grid container direction='row' justify='space-between' spacing={24}>
<Grid item xs={6}>
<Table>
<TableHead>
<TableRow>
<TableCell>IP</TableCell>
<TableCell>Location</TableCell>
<TableCell>Node ID</TableCell>
<TableCell>Location</TableCell>
<TableCell>Traffic</TableCell>
<TableCell>Connected</TableCell>
<TableCell>Disconnected</TableCell>
</TableRow>
</TableHead>
<TableBody>
@ -171,22 +170,21 @@ class Network extends Component<Props, State> {
return null;
}
return (
<TableRow key={`known${ip}`}>
<TableCell>{ip}</TableCell>
<TableRow key={`known_${ip}`}>
<TableCell>
{Object.keys(bundle.knownPeers).map(id => id.substring(0, 10)).join(' ')}
</TableCell>
<TableCell>
{bundle.location ? (() => {
const l = bundle.location;
return `${l.country ? l.country : ''}${l.city ? `/${l.city}` : ''} ${l.latitude} ${l.longitude}`;
return `${l.country ? l.country : ''}${l.city ? `/${l.city}` : ''}`;
})() : ''}
</TableCell>
<TableCell>
{Object.keys(bundle.knownPeers).map(id => id.substring(0, 10)).join(' ')}
</TableCell>
<TableCell>
{Object.values(bundle.knownPeers).map(({ingress, egress}) => (
<div>
<AreaChart
width={300} height={50}
width={200} height={18}
syncId={'footerSyncId'}
data={egress.map(({value}) => ({egress: value || 0}))}
margin={{top: 5, right: 5, bottom: 0, left: 5}}
@ -195,7 +193,7 @@ class Network extends Component<Props, State> {
<Area isAnimationActive={false} type='monotone' dataKey='egress' stroke='#8884d8' fill='#8884d8' />
</AreaChart>
<AreaChart
width={300} height={50}
width={200} height={18}
syncId={'footerSyncId'}
data={ingress.map(({value}) => ({ingress: -value || 0}))}
margin={{top: 0, right: 5, bottom: 5, left: 5}}
@ -206,52 +204,46 @@ class Network extends Component<Props, State> {
</div>
))}
</TableCell>
<TableCell>
{Object.values(bundle.knownPeers).map(peer => peer.connected && peer.connected.map(time => this.formatTime(time)).join(' ')).join(', ')}
</TableCell>
<TableCell>
{Object.values(bundle.knownPeers).map(peer => peer.disconnected && peer.disconnected.map(time => this.formatTime(time)).join(' ')).join(', ')}
</TableCell>
</TableRow>
);
})}
</TableBody>
</Table>
</Grid>
<Grid item xs={6}>
<Table>
<TableHead>
<TableRow>
<TableCell>IP</TableCell>
<TableCell>Location</TableCell>
<TableCell>Connected</TableCell>
<TableCell>Disconnected</TableCell>
<TableCell>Attempts</TableCell>
</TableRow>
</TableHead>
<TableBody>
{Object.entries(this.props.content.peers.bundles).map(([ip, bundle]) => {
if (!bundle.unknownPeers || bundle.unknownPeers.length < 1) {
if (!bundle.attempts || bundle.attempts.length < 1) {
return null;
}
return (
<TableRow key={`unknown${ip}`}>
<TableRow key={`attempt_${ip}`}>
<TableCell>{ip}</TableCell>
<TableCell>
{bundle.location ? (() => {
const l = bundle.location;
return `${l.country ? l.country : ''}${l.city ? `/${l.city}` : ''} ${l.latitude} ${l.longitude}`;
return `${l.country ? l.country : ''}${l.city ? `/${l.city}` : ''}`;
})() : ''}
</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(', ')}
{Object.values(bundle.attempts).length}
</TableCell>
</TableRow>
);
})}
</TableBody>
</Table>
</div>);
</Grid>
</Grid>
);
}
}

3
dashboard/assets/index.jsx
View file

@ -25,6 +25,9 @@ import createMuiTheme from '@material-ui/core/styles/createMuiTheme';
import Dashboard from './components/Dashboard';
const theme: Object = createMuiTheme({
// typography: {
// useNextVariants: true,
// },
palette: {
type: 'dark',
},

48
dashboard/assets/package.json
View file

@ -1,48 +1,48 @@
{
"private": true,
"dependencies": {
"@babel/core": "7.1.2",
"@babel/plugin-proposal-class-properties": "7.1.0",
"@babel/plugin-proposal-function-bind": "^7.0.0",
"@babel/plugin-transform-flow-strip-types": "^7.0.0",
"@babel/preset-env": "7.1.0",
"@babel/core": "^7.2.2",
"@babel/plugin-proposal-class-properties": "7.2.3",
"@babel/plugin-proposal-function-bind": "7.2.0",
"@babel/plugin-transform-flow-strip-types": "7.2.3",
"@babel/preset-env": "7.2.3",
"@babel/preset-react": "^7.0.0",
"@babel/preset-stage-0": "^7.0.0",
"@material-ui/core": "3.2.0",
"@material-ui/core": "3.8.1",
"@material-ui/icons": "^3.0.1",
"babel-eslint": "10.0.1",
"babel-loader": "8.0.4",
"babel-loader": "8.0.5",
"classnames": "^2.2.6",
"css-loader": "^1.0.0",
"css-loader": "2.1.0",
"escape-html": "^1.0.3",
"eslint": "5.7.0",
"eslint": "5.11.1",
"eslint-config-airbnb": "^17.0.0",
"eslint-loader": "2.1.1",
"eslint-plugin-flowtype": "3.0.0",
"eslint-plugin-flowtype": "3.2.0",
"eslint-plugin-import": "^2.13.0",
"eslint-plugin-jsx-a11y": "6.1.2",
"eslint-plugin-node": "^7.0.1",
"eslint-plugin-node": "8.0.0",
"eslint-plugin-promise": "4.0.1",
"eslint-plugin-react": "7.11.1",
"file-loader": "2.0.0",
"flow-bin": "0.83.0",
"eslint-plugin-react": "7.12.2",
"file-loader": "3.0.1",
"flow-bin": "0.89.0",
"flow-bin-loader": "^1.0.3",
"flow-typed": "^2.5.1",
"path": "^0.12.7",
"react": "16.5.2",
"react-dom": "16.5.2",
"react": "16.7.0",
"react-dom": "16.7.0",
"react-fa": "^5.0.0",
"react-hot-loader": "4.3.11",
"react-transition-group": "2.5.0",
"recharts": "1.3.4",
"react-hot-loader": "4.6.3",
"react-transition-group": "2.5.2",
"recharts": "1.4.2",
"style-loader": "0.23.1",
"uglifyjs-webpack-plugin": "2.0.1",
"uglifyjs-webpack-plugin": "2.1.1",
"url": "^0.11.0",
"url-loader": "1.1.2",
"webpack": "4.20.2",
"webpack-cli": "3.1.2",
"webpack-dev-server": "3.1.9",
"webpack-merge": "^4.1.4"
"webpack": "4.28.3",
"webpack-cli": "3.2.0",
"webpack-dev-server": "3.1.14",
"webpack-merge": "4.1.5"
},
"scripts": {
"build": "webpack --config webpack.config.prod.js",

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

@ -29,7 +29,6 @@ export type Content = {
export type ChartEntries = Array<ChartEntry>;
export type ChartEntry = {
time: Date,
value: number,
};
@ -73,7 +72,7 @@ export type Peers = {
export type PeerBundle = {
location: GeoLocation,
knownPeers: {[string]: KnownPeer},
unknownPeers: Array<UnknownPeer>,
attempts: Array<UnknownPeer>,
};
export type KnownPeer = {

1626
dashboard/assets/yarn.lock
View file

File diff suppressed because it is too large Load diff

24
dashboard/dashboard.go
View file

@ -60,7 +60,7 @@ type Dashboard struct {
peerLock sync.RWMutex // Lock protecting the stored peer data
logLock sync.RWMutex // Lock protecting the stored log data
geodb *GeoDB // geoip database instance for IP to geographical information conversions
geodb *geoDB // geoip database instance for IP to geographical information conversions
logdir string // Directory containing the log files
quit chan chan error // Channel used for graceful exit
@ -91,14 +91,14 @@ func New(config *Config, commit string, logdir string) *Dashboard {
Version: fmt.Sprintf("v%d.%d.%d%s", params.VersionMajor, params.VersionMinor, params.VersionPatch, versionMeta),
},
System: &SystemMessage{
ActiveMemory: emptyChartEntries(now, sampleLimit, config.Refresh),
VirtualMemory: emptyChartEntries(now, sampleLimit, config.Refresh),
NetworkIngress: emptyChartEntries(now, sampleLimit, config.Refresh),
NetworkEgress: emptyChartEntries(now, sampleLimit, config.Refresh),
ProcessCPU: emptyChartEntries(now, sampleLimit, config.Refresh),
SystemCPU: emptyChartEntries(now, sampleLimit, config.Refresh),
DiskRead: emptyChartEntries(now, sampleLimit, config.Refresh),
DiskWrite: emptyChartEntries(now, sampleLimit, config.Refresh),
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),
},
},
logdir: logdir,
@ -106,12 +106,10 @@ func New(config *Config, commit string, logdir string) *Dashboard {
}
// emptyChartEntries returns a ChartEntry array containing limit number of empty samples.
func emptyChartEntries(t time.Time, limit int, refresh time.Duration) ChartEntries {
func emptyChartEntries(t time.Time, limit int) ChartEntries {
ce := make(ChartEntries, limit)
for i := 0; i < limit; i++ {
ce[i] = &ChartEntry{
Time: t.Add(-time.Duration(i) * refresh),
}
ce[i] = new(ChartEntry)
}
return ce
}

28
dashboard/geoip.go
View file

@ -23,9 +23,9 @@ import (
"github.com/apilayer/freegeoip"
)
// GeoDBInfo contains all the geographical information we could extract based on an IP
// geoDBInfo contains all the geographical information we could extract based on an IP
// address.
type GeoDBInfo struct {
type geoDBInfo struct {
Country struct {
Names struct {
English string `maxminddb:"en" json:"en,omitempty"`
@ -42,22 +42,22 @@ type GeoDBInfo struct {
} `maxminddb:"location" json:"location,omitempty"`
}
// GeoLocation contains geographical information.
type GeoLocation struct {
// geoLocation contains geographical information.
type geoLocation struct {
Country string `json:"country,omitempty"`
City string `json:"city,omitempty"`
Latitude float64 `json:"latitude,omitempty"`
Longitude float64 `json:"longitude,omitempty"`
}
// GeoDB represents a geoip database that can be queried for IP to geographical
// geoDB represents a geoip database that can be queried for IP to geographical
// information conversions.
type GeoDB struct {
type geoDB struct {
geodb *freegeoip.DB
}
// Open creates a new geoip database with an up-to-date database from the internet.
func OpenGeoDB() (*GeoDB, error) {
func openGeoDB() (*geoDB, error) {
// Initiate a geoip database to cross reference locations
db, err := freegeoip.OpenURL(freegeoip.MaxMindDB, 24*time.Hour, time.Hour)
if err != nil {
@ -70,26 +70,26 @@ func OpenGeoDB() (*GeoDB, error) {
return nil, err
}
// Assemble and return our custom wrapper
return &GeoDB{geodb: db}, nil
return &geoDB{geodb: db}, nil
}
// Close terminates the database background updater.
func (db *GeoDB) Close() error {
func (db *geoDB) close() error {
db.geodb.Close()
return nil
}
// Lookup converts an IP address to a geographical location.
func (db *GeoDB) Lookup(ip net.IP) *GeoDBInfo {
result := new(GeoDBInfo)
func (db *geoDB) lookup(ip net.IP) *geoDBInfo {
result := new(geoDBInfo)
db.geodb.Lookup(ip, result)
return result
}
// Location retrieves the geographical location of the given IP address.
func (db *GeoDB) Location(ip string) *GeoLocation {
location := db.Lookup(net.ParseIP(ip))
return &GeoLocation{
func (db *geoDB) location(ip string) *geoLocation {
location := db.lookup(net.ParseIP(ip))
return &geoLocation{
Country: location.Country.Names.English,
City: location.City.Names.English,
Latitude: location.Location.Latitude,

6
dashboard/message.go
View file

@ -18,7 +18,6 @@ package dashboard
import (
"encoding/json"
"time"
)
type Message struct {
@ -34,7 +33,6 @@ type Message struct {
type ChartEntries []*ChartEntry
type ChartEntry struct {
Time time.Time `json:"time"`
Value float64 `json:"value"`
}
@ -58,8 +56,8 @@ type TxPoolMessage struct {
// NetworkMessage contains information about the peers
// organized based on their IP address and node ID.
type NetworkMessage struct {
Peers *PeerContainer `json:"peers,omitempty"` // Peer tree.
Diff []*PeerEvent `json:"diff,omitempty"` // Events that change the peer tree.
Peers *peerContainer `json:"peers,omitempty"` // Peer tree.
Diff []*peerEvent `json:"diff,omitempty"` // Events that change the peer tree.
}
// SystemMessage contains the metered system data samples.

289
dashboard/peers.go
View file

@ -28,9 +28,9 @@ import (
)
const (
eventBufferLimit = 128 // Maximum number of events of buffered peer events.
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.
eventBufferLimit = 128 // Maximum number of buffered peer events.
knownPeerLimit = 100 // Maximum number of stored peers, which successfully made the handshake.
attemptLimit = 100 // 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
@ -39,16 +39,17 @@ const (
eventLimit = knownPeerLimit << 2
)
// PeerContainer contains information about the node's peers. This data structure
// peerContainer contains information about the node's peers. This data structure
// maintains the metered peer data based on the different behaviours of the peers.
//
// Every peer has an IP address, and the peers that manage to make the handshake
// (known peers) have node IDs too. There can appear more peers with the same IP,
// therefore the peer maintainer data structure is a tree consisting of a map of
// therefore the peer container data structure is a tree consisting of a map of
// maps, where the first key groups the peers by IP, while the second one groups
// them by the node ID. The peers failing before the handshake (unknown peers)
// only have IP addresses, so their connection attempts are stored as part of the
// value of the outer map.
// them by the node ID. The known peers can be active if their connection is still
// open, or inactive otherwise. The peers failing before the handshake (unknown
// peers) only have IP addresses, so their connection attempts are stored as part
// of the value of the outer map.
//
// Another criteria is to limit the number of metered peers so that
// they don't fill the memory. The selection order is based on the
@ -61,12 +62,9 @@ const (
// the limit, the first element is removed from the list, as well as from
// the tree.
//
// The active peers that are still connected have priority over the disconnected
// ones, therefore the list is extended by a separator, which is a pointer to a
// list element. The separator separates the active peers from the inactive ones,
// and it is the entry for the list. If the peer that is to be inserted is active,
// it goes after the separator, otherwise it goes before. This way the active peers
// never move to the front before the inactive ones.
// The active peers have priority over the inactive ones, therefore
// they have their own list. The separation makes it sure that the
// inactive peers are always removed before the active ones.
//
// The peers that don't manage to make handshake are not inserted into the list,
// only their connection attempts are appended to the array belonging to their IP.
@ -76,116 +74,99 @@ const (
//
// This data structure makes it possible to marshal the peer
// history simply by passing it to the JSON marshaler.
type PeerContainer struct {
type peerContainer struct {
// Bundles is the outer map using the peer's IP address as key.
Bundles map[string]*PeerBundle `json:"bundles,omitempty"`
Bundles map[string]*peerBundle `json:"bundles,omitempty"`
// activeSeparator is a pointer to the last inactive peer element, splitting
// the list into an inactive and an active part, and forming the entry for
// the peer list.
activeSeparator *list.Element
// activePeers contains the peers with opened connection in random order.
activePeers *list.List
// knownPeers contains the peers that managed to make handshake.
knownPeers *list.List
// inactivePeers contains the peers with closed connection in chronological order.
inactivePeers *list.List
// unknownPeers is the super array containing the IP addresses, from which
// 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.
unknownPeers []string
attemptOrder []string
// geodb is the geoip database used to retrieve the peers' geographical location.
geodb *GeoDB
// refresh is the refresh rate used to generate the
// initial auxiliary traffic samples' time stamps.
refresh time.Duration
geodb *geoDB
}
// NewPeerContainer returns a new instance of the peer container.
func NewPeerContainer(geodb *GeoDB, refresh time.Duration) *PeerContainer {
return &PeerContainer{
Bundles: make(map[string]*PeerBundle),
knownPeers: list.New(),
unknownPeers: make([]string, 0, unknownPeerLimit),
// newPeerContainer returns a new instance of the peer container.
func newPeerContainer(geodb *geoDB) *peerContainer {
return &peerContainer{
Bundles: make(map[string]*peerBundle),
activePeers: list.New(),
inactivePeers: list.New(),
attemptOrder: make([]string, 0, attemptLimit),
geodb: geodb,
refresh: refresh,
}
}
// getOrInitBundle inserts a new peer bundle into the map, if the peer belonging
// bundle inserts a new peer bundle into the map, if the peer belonging
// to the given IP wasn't metered so far. In this case retrieves the location of
// 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) getOrInitBundle(ip string) (*PeerBundle, []*PeerEvent) {
var events []*PeerEvent
func (pc *peerContainer) bundle(ip string) (*peerBundle, []*peerEvent) {
var events []*peerEvent
if _, ok := pc.Bundles[ip]; !ok {
location := pc.geodb.Location(ip)
events = append(events, &PeerEvent{
location := pc.geodb.location(ip)
events = append(events, &peerEvent{
IP: ip,
Location: location,
})
pc.Bundles[ip] = &PeerBundle{
pc.Bundles[ip] = &peerBundle{
Location: location,
KnownPeers: make(map[string]*KnownPeer),
KnownPeers: make(map[string]*knownPeer),
}
}
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) []*PeerEvent {
peer, events := pc.getOrInitKnownPeer(event.IP, event.ID)
func (pc *peerContainer) extendKnown(event *peerEvent) []*peerEvent {
bundle, events := pc.bundle(event.IP)
peer, peerEvents := bundle.knownPeer(event.IP, event.ID)
events = append(events, peerEvents...)
// Append the connect and the disconnect events to
// the corresponding arrays keeping the limit.
if event.Connected != nil {
switch {
case event.Connected != nil:
peer.Connected = append(peer.Connected, event.Connected)
if first := len(peer.Connected) - sampleLimit; first > 0 {
peer.Connected = peer.Connected[first:]
}
}
if event.Disconnected != nil {
case event.Disconnected != nil:
peer.Disconnected = append(peer.Disconnected, event.Disconnected)
if first := len(peer.Disconnected) - sampleLimit; first > 0 {
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)
_ = pc.activePeers.Remove(peer.listElement)
_ = pc.inactivePeers.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
peer.listElement = pc.activePeers.PushBack(peer)
for pc.activePeers.Len()+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)...)
} else {
log.Warn("Failed to insert known peer", "peer", *peer)
log.Warn("Failed to parse the removed 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 {
for pc.activePeers.Len() > knownPeerLimit {
if removedPeer, ok := pc.activePeers.Remove(pc.activePeers.Front()).(*knownPeer); ok {
events = append(events, pc.removeKnown(removedPeer.ip, removedPeer.id)...)
} else {
log.Warn("Failed to parse the removed peer")
@ -194,140 +175,125 @@ func (pc *PeerContainer) extendKnown(event *PeerEvent) []*PeerEvent {
return events
}
// extendUnknown handles the events of the peers failing before/during the handshake.
// handleAttempt handles the events of the peers failing before/during the handshake.
// Returns the events occurring during the extension.
func (pc *PeerContainer) extendUnknown(event *PeerEvent) []*PeerEvent {
bundle, events := pc.getOrInitBundle(event.IP)
bundle.UnknownPeers = append(bundle.UnknownPeers, &UnknownPeer{
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.unknownPeers = append(pc.unknownPeers, event.IP)
for len(pc.unknownPeers) > unknownPeerLimit {
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.removeUnknown(pc.unknownPeers[0])...)
pc.unknownPeers = pc.unknownPeers[1:]
events = append(events, pc.removeAttempt(pc.attemptOrder[0])...)
pc.attemptOrder = pc.attemptOrder[1:]
}
return events
}
// setActive moves the peer denoted by the given IP address and node ID after
// the list's active separator. Takes no effect if the peer doesn't exist.
func (pc *PeerContainer) setActive(ip, id string) {
// setActive pushes the peer denoted by the given IP address and node ID
// into the active peer list. 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 {
if peer.listElement != nil {
// 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.
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 {
log.Warn("Failed to insert the peer after the separator", "peer", peer)
_ = pc.inactivePeers.Remove(peer.listElement)
_ = pc.activePeers.Remove(peer.listElement)
}
peer.listElement = pc.activePeers.PushBack(peer)
}
}
}
// resetActiveSeparator resets the active separator, denoting
// that active peers are not considered active anymore.
func (pc *PeerContainer) resetActiveSeparator() {
pc.activeSeparator = nil
// resetActive pushes the active peers to the end of the inactive peer
// list, denoting that active peers are not considered active anymore.
func (pc *peerContainer) resetActive() {
for pc.activePeers.Front() != nil {
pc.inactivePeers.PushBack(pc.activePeers.Remove(pc.activePeers.Front()))
}
}
// PeerBundle contains the peers belonging to a given IP address.
type PeerBundle struct {
// 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.
Location *GeoLocation `json:"location,omitempty"`
Location *geoLocation `json:"location,omitempty"`
// KnownPeers is the inner map of the metered peer
// maintainer data structure using the node ID as key.
KnownPeers map[string]*KnownPeer `json:"knownPeers,omitempty"`
KnownPeers map[string]*knownPeer `json:"knownPeers,omitempty"`
// UnknownPeers contains the failed connection attempts of the
// Attempts contains the failed connection attempts of the
// peers belonging to a given IP address in chronological order.
UnknownPeers []*UnknownPeer `json:"unknownPeers,omitempty"`
Attempts []*peerAttempt `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(ip, id string) (events []*peerEvent) {
if bundle, ok := pc.Bundles[ip]; ok {
if _, ok := bundle.KnownPeers[id]; ok {
events = append(events, &PeerEvent{
events = append(events, &peerEvent{
Remove: RemoveKnown,
IP: ip,
ID: id,
})
delete(bundle.KnownPeers, id)
} else {
log.Warn("No peer to remove", ip, id)
log.Warn("No peer to remove", "ip", ip, "id", id)
}
if len(bundle.KnownPeers) < 1 && len(bundle.UnknownPeers) < 1 {
events = append(events, &PeerEvent{
if len(bundle.KnownPeers) < 1 && len(bundle.Attempts) < 1 {
events = append(events, &peerEvent{
Remove: RemoveBundle,
IP: ip,
})
delete(pc.Bundles, ip)
}
} else {
log.Warn("No bundle to remove", ip)
}
return events
}
// removeUnknown removes the unknown peer belonging to the
// removeAttempt removes the peer attempt belonging to the
// given IP address and node ID from the peer tree.
func (pc *PeerContainer) removeUnknown(ip string) (events []*PeerEvent) {
func (pc *peerContainer) removeAttempt(ip string) (events []*peerEvent) {
if bundle, ok := pc.Bundles[ip]; ok {
if len(bundle.UnknownPeers) > 0 {
events = append(events, &PeerEvent{
Remove: RemoveUnknown,
if len(bundle.Attempts) > 0 {
events = append(events, &peerEvent{
Remove: RemoveAttempt,
IP: ip,
})
bundle.UnknownPeers = bundle.UnknownPeers[1:]
bundle.Attempts = bundle.Attempts[1:]
}
if len(bundle.UnknownPeers) < 1 && len(bundle.KnownPeers) < 1 {
events = append(events, &PeerEvent{
if len(bundle.Attempts) < 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 events
}
// getOrInitKnownPeer inserts a new peer into the map, if the peer belonging
// knownPeer inserts a new peer into the map, if the peer belonging
// 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) getOrInitKnownPeer(ip, id string, refresh time.Duration) (*KnownPeer, []*PeerEvent) {
var events []*PeerEvent
func (bundle *peerBundle) knownPeer(ip, id string) (*knownPeer, []*peerEvent) {
var events []*peerEvent
if _, ok := bundle.KnownPeers[id]; !ok {
now := time.Now()
ingress := emptyChartEntries(now, sampleLimit, refresh)
egress := emptyChartEntries(now, sampleLimit, refresh)
events = append(events, &PeerEvent{
ingress := emptyChartEntries(now, sampleLimit)
egress := emptyChartEntries(now, sampleLimit)
events = append(events, &peerEvent{
IP: ip,
ID: id,
Ingress: append([]*ChartEntry{}, ingress...),
Egress: append([]*ChartEntry{}, egress...),
})
bundle.KnownPeers[id] = &KnownPeer{
bundle.KnownPeers[id] = &knownPeer{
ip: ip,
id: id,
Ingress: ingress,
@ -337,8 +303,8 @@ func (bundle *PeerBundle) getOrInitKnownPeer(ip, id string, refresh time.Duratio
return bundle.KnownPeers[id], events
}
// KnownPeer contains the metered data of a particular peer.
type KnownPeer struct {
// knownPeer contains the metered data of a particular peer.
type knownPeer struct {
// Connected contains the timestamps of the peer's connection events.
Connected []*time.Time `json:"connected,omitempty"`
@ -361,8 +327,8 @@ type KnownPeer struct {
prevEgress float64
}
// UnknownPeer contains a failed peer connection attempt's attributes.
type UnknownPeer struct {
// 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"`
@ -375,25 +341,27 @@ type RemovedPeerType string
const (
RemoveKnown RemovedPeerType = "known"
RemoveUnknown RemovedPeerType = "unknown"
RemoveAttempt RemovedPeerType = "attempt"
RemoveBundle RemovedPeerType = "bundle"
)
// PeerEvent contains the attributes of a peer event.
type PeerEvent struct {
// 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.
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.
}
// trafficMap
// trafficMap is a container for the periodically collected peer traffic.
type trafficMap map[string]map[string]float64
// insert inserts a new value to the traffic map. Overwrites
// the value at the given ip and id if that already exists.
func (m *trafficMap) insert(ip, id string, val float64) {
if _, ok := (*m)[ip]; !ok {
(*m)[ip] = make(map[string]float64)
@ -407,12 +375,12 @@ func (db *Dashboard) collectPeerData() {
// Open the geodb database for IP to geographical information conversions.
var err error
db.geodb, err = OpenGeoDB()
db.geodb, err = openGeoDB()
if err != nil {
log.Warn("Failed to open geodb", "err", err)
return
}
defer db.geodb.Close()
defer db.geodb.close()
peerCh := make(chan p2p.MeteredPeerEvent, eventBufferLimit) // Peer event channel.
subPeer := p2p.SubscribeMeteredPeerEvent(peerCh) // Subscribe to peer events.
@ -449,7 +417,7 @@ func (db *Dashboard) collectPeerData() {
collectIngress := trafficCollector(p2p.MetricsInboundTraffic + "/")
collectEgress := trafficCollector(p2p.MetricsOutboundTraffic + "/")
peers := NewPeerContainer(db.geodb, db.config.Refresh)
peers := newPeerContainer(db.geodb)
db.peerLock.Lock()
db.history.Network = &NetworkMessage{
Peers: peers,
@ -458,7 +426,7 @@ func (db *Dashboard) collectPeerData() {
// newPeerEvents contains peer events, which trigger operations that
// will be executed on the peer tree after a metering period.
newPeerEvents := make([]*PeerEvent, 0, eventLimit)
newPeerEvents := make([]*peerEvent, 0, eventLimit)
ingress, egress := new(trafficMap), new(trafficMap)
*ingress, *egress = make(trafficMap), make(trafficMap)
@ -469,14 +437,14 @@ func (db *Dashboard) collectPeerData() {
switch event.Type {
case p2p.PeerConnected:
connected := now.Add(-event.Elapsed)
newPeerEvents = append(newPeerEvents, &PeerEvent{
newPeerEvents = append(newPeerEvents, &peerEvent{
IP: event.IP.String(),
ID: event.ID.String(),
Connected: &connected,
})
case p2p.PeerDisconnected:
ip, id := event.IP.String(), event.ID.String()
newPeerEvents = append(newPeerEvents, &PeerEvent{
newPeerEvents = append(newPeerEvents, &peerEvent{
IP: ip,
ID: id,
Disconnected: &now,
@ -491,7 +459,7 @@ func (db *Dashboard) collectPeerData() {
egress.insert(ip, id, float64(event.Egress))
case p2p.PeerHandshakeFailed:
connected := now.Add(-event.Elapsed)
newPeerEvents = append(newPeerEvents, &PeerEvent{
newPeerEvents = append(newPeerEvents, &peerEvent{
IP: event.IP.String(),
Connected: &connected,
Disconnected: &now,
@ -517,9 +485,9 @@ func (db *Dashboard) collectPeerData() {
// the diff, otherwise the active peers can be removed.
//
// After a metering period the active peers can become inactive,
// so resetting the separator makes it sure, that only the active
// peers move to the protected part of the list.
peers.resetActiveSeparator()
// so at the beginning it is necessary to transpose them to the
// inactive list.
peers.resetActive()
for ip, bundle := range *ingress {
for id := range bundle {
// Only set the peers that are inserted both
@ -531,7 +499,7 @@ func (db *Dashboard) collectPeerData() {
}
}
}
var diff []*PeerEvent
var diff []*peerEvent
for i := 0; i < len(newPeerEvents); i++ {
if newPeerEvents[i].IP == "" {
log.Warn("Peer event without IP", "event", *newPeerEvents[i])
@ -541,18 +509,16 @@ func (db *Dashboard) collectPeerData() {
// 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.
// to an unknown one, which is considered as connection attempt.
//
// 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])...)
diff = append(diff, peers.handleAttempt(newPeerEvents[i])...)
continue
}
diff = append(diff, peers.extendKnown(newPeerEvents[i])...)
}
now := time.Now()
// Update the peer tree using the traffic maps.
for ip, bundle := range peers.Bundles {
for id, peer := range bundle.KnownPeers {
@ -568,11 +534,9 @@ func (db *Dashboard) collectPeerData() {
}
peer.prevIngress, peer.prevEgress = curIngress, curEgress
i := &ChartEntry{
Time: now,
Value: deltaIngress,
}
e := &ChartEntry{
Time: now,
Value: deltaEgress,
}
peer.Ingress = append(peer.Ingress, i)
@ -584,7 +548,7 @@ func (db *Dashboard) collectPeerData() {
peer.Egress = peer.Egress[first:]
}
// Creating the traffic sample events.
diff = append(diff, &PeerEvent{
diff = append(diff, &peerEvent{
IP: ip,
ID: id,
Ingress: ChartEntries{i},
@ -592,13 +556,8 @@ 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: diff,
}})

10
dashboard/system.go
View file

@ -92,39 +92,29 @@ func (db *Dashboard) collectSystemData() {
prevDiskRead = curDiskRead
prevDiskWrite = curDiskWrite
now := time.Now()
runtime.ReadMemStats(&mem)
activeMemory := &ChartEntry{
Time: now,
Value: float64(mem.Alloc) / frequency,
}
virtualMemory := &ChartEntry{
Time: now,
Value: float64(mem.Sys) / frequency,
}
networkIngress := &ChartEntry{
Time: now,
Value: deltaNetworkIngress / frequency,
}
networkEgress := &ChartEntry{
Time: now,
Value: deltaNetworkEgress / frequency,
}
processCPU := &ChartEntry{
Time: now,
Value: deltaProcessCPUTime / frequency / numCPU * 100,
}
systemCPU := &ChartEntry{
Time: now,
Value: float64(deltaSystemCPUUsage.Sys+deltaSystemCPUUsage.User) / frequency / numCPU,
}
diskRead := &ChartEntry{
Time: now,
Value: float64(deltaDiskRead) / frequency,
}
diskWrite := &ChartEntry{
Time: now,
Value: float64(deltaDiskWrite) / frequency,
}
db.sysLock.Lock()

1
p2p/metrics.go
View file

@ -161,6 +161,7 @@ func (c *meteredConn) Write(b []byte) (n int, err error) {
// 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)
if atomic.AddInt32(&meteredPeerCount, 1) >= MeteredPeerLimit {
// Don't register the peer in the traffic registries.
atomic.AddInt32(&meteredPeerCount, -1)