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

dashboard, p2p: use metrics registry with event feeds

Browse source
This commit is contained in:
Kurkó Mihály 2018-09-12 14:13:04 +03:00
parent c183fa4137
commit 80135738a6
9 changed files with 287 additions and 461 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
dashboard/assets/components/SideBar.jsx
View file

@ -51,6 +51,7 @@ const themeStyles = theme => ({
},
icon: {
fontSize: theme.spacing.unit * 3,
overflow: 'unset',
},
});

123
dashboard/assets/webpack.config.js
View file

@ -1,123 +0,0 @@
// Copyright 2017 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
const webpack = require('webpack');
const UglifyJsPlugin = require('uglifyjs-webpack-plugin');
const path = require('path');
module.exports = {
mode: 'development',
target: 'web',
entry: {
bundle: './index',
},
output: {
filename: '[name].js',
path: path.resolve(__dirname, ''),
// sourceMapFilename: '[file].map',
},
resolve: {
modules: [
'node_modules',
path.resolve(__dirname, 'components'), // import './components/Component' -> import 'Component'
],
// alias: {
// root: path.resolve(__dirname, ''),
// },
extensions: ['.js', '.jsx'],
},
devtool: 'eval',
// devtool: 'inline-source-map',
optimization: {
minimize: true,
namedModules: true, // Module names instead of numbers - resolves the large diff problem.
minimizer: [
new UglifyJsPlugin({
uglifyOptions: {
compress: true,
mangle: true,
output: {
comments: false,
beautify: true,
bracketize: true,
},
warnings: true,
},
// sourceMap: true,
}),
],
},
plugins: [
new webpack.DefinePlugin({
PROD: process.env.NODE_ENV === 'production',
}),
new webpack.HotModuleReplacementPlugin(),
],
module: {
rules: [
{
test: /\.jsx$/, // regexp for JSX files
exclude: /node_modules/,
use: [ // order: from bottom to top
{
loader: 'babel-loader',
options: {
plugins: [ // order: from top to bottom
'@babel/proposal-function-bind', // instead of stage 0
'@babel/proposal-class-properties', // static defaultProps
'@babel/transform-flow-strip-types',
'react-hot-loader/babel',
],
presets: [ // order: from bottom to top
'@babel/env',
'@babel/react',
],
},
},
// 'eslint-loader', // show errors in the console
],
},
{
test: /\.css$/,
oneOf: [
{
test: /font-awesome/,
use: [
'style-loader',
'css-loader',
path.resolve(__dirname, './fa-only-woff-loader.js'),
],
},
{
use: [
'style-loader',
'css-loader',
],
},
],
},
{
test: /\.woff2?$/, // font-awesome icons
use: 'url-loader',
},
],
},
devServer: {
port: 8081,
hot: true,
compress: true,
},
};

27
dashboard/dashboard.go
View file

@ -65,12 +65,9 @@ type Dashboard struct {
conns map[uint32]*client // Currently live websocket connections
nextConnID uint32 // Next connection id
history *Message // Stored general data
sysHistory *SystemMessage // Stored system data
networkHistory *NetworkMessage // Stored peer data
logHistory *LogsMessage // Stored log data
history *Message // Stored historical data
lock sync.RWMutex // Lock protecting the dashboard's internals
lock sync.Mutex // Lock protecting the dashboard's internals
sysLock sync.RWMutex // Lock protecting the stored system data
peerLock sync.RWMutex // Lock protecting the stored peer data
logLock sync.RWMutex // Lock protecting the stored log data
@ -105,8 +102,7 @@ func New(config *Config, commit string, logdir string) *Dashboard {
Commit: commit,
Version: fmt.Sprintf("v%d.%d.%d%s", params.VersionMajor, params.VersionMinor, params.VersionPatch, versionMeta),
},
},
sysHistory: &SystemMessage{
System: &SystemMessage{
ActiveMemory: emptyChartEntries(now, activeMemorySampleLimit, config.Refresh),
VirtualMemory: emptyChartEntries(now, virtualMemorySampleLimit, config.Refresh),
NetworkIngress: emptyChartEntries(now, networkIngressSampleLimit, config.Refresh),
@ -116,9 +112,10 @@ func New(config *Config, commit string, logdir string) *Dashboard {
DiskRead: emptyChartEntries(now, diskReadSampleLimit, config.Refresh),
DiskWrite: emptyChartEntries(now, diskWriteSampleLimit, config.Refresh),
},
networkHistory: &NetworkMessage{
Network: &NetworkMessage{
PeerBundles: make(map[string]*PeerBundle),
},
},
logdir: logdir,
}
}
@ -248,18 +245,16 @@ func (db *Dashboard) apiHandler(conn *websocket.Conn) {
}()
// Send the past data.
db.lock.RLock()
h := deepcopy.Copy(db.history).(*Message)
db.lock.RUnlock()
db.sysLock.RLock()
h.System = deepcopy.Copy(db.sysHistory).(*SystemMessage)
db.sysLock.RUnlock()
db.peerLock.RLock()
h.Network = deepcopy.Copy(db.networkHistory).(*NetworkMessage)
db.peerLock.RUnlock()
db.logLock.RLock()
h.Logs = deepcopy.Copy(db.logHistory).(*LogsMessage)
h := deepcopy.Copy(db.history).(*Message)
db.sysLock.RUnlock()
db.peerLock.RUnlock()
db.logLock.RUnlock()
client.msg <- h
// Start tracking the connection and drop at connection loss.

26
dashboard/log.go
View file

@ -95,9 +95,9 @@ func (db *Dashboard) handleLogRequest(r *LogsRequest, c *client) {
// so in order to avoid log record duplication on the client side, it is
// handled differently. Its actual content is always saved in the history.
db.logLock.RLock()
if db.logHistory != nil {
if db.history.Logs != nil {
c.msg <- &Message{
Logs: deepcopy.Copy(db.logHistory).(*LogsMessage),
Logs: deepcopy.Copy(db.history.Logs).(*LogsMessage),
}
}
db.logLock.RUnlock()
@ -175,7 +175,7 @@ func (db *Dashboard) streamLogs() {
return
}
db.logLock.Lock()
db.logHistory = &LogsMessage{
db.history.Logs = &LogsMessage{
Source: &LogFile{
Name: fi.Name(),
Last: true,
@ -241,8 +241,8 @@ loop:
break loop
}
db.logLock.Lock()
db.logHistory.Source.Name = fi.Name()
db.logHistory.Chunk = emptyChunk
db.history.Logs.Source.Name = fi.Name()
db.history.Logs.Chunk = emptyChunk
db.logLock.Unlock()
case <-ticker.C: // Send log updates to the client.
if opened == nil {
@ -267,15 +267,15 @@ loop:
var l *LogsMessage
// Update the history.
db.logLock.Lock()
if bytes.Equal(db.logHistory.Chunk, emptyChunk) {
db.logHistory.Chunk = chunk
l = deepcopy.Copy(db.logHistory).(*LogsMessage)
if bytes.Equal(db.history.Logs.Chunk, emptyChunk) {
db.history.Logs.Chunk = chunk
l = deepcopy.Copy(db.history.Logs).(*LogsMessage)
} else {
b = make([]byte, len(db.logHistory.Chunk)+len(chunk)-1)
copy(b, db.logHistory.Chunk)
b[len(db.logHistory.Chunk)-1] = ','
copy(b[len(db.logHistory.Chunk):], chunk[1:])
db.logHistory.Chunk = b
b = make([]byte, len(db.history.Logs.Chunk)+len(chunk)-1)
copy(b, db.history.Logs.Chunk)
b[len(db.history.Logs.Chunk)-1] = ','
copy(b[len(db.history.Logs.Chunk):], chunk[1:])
db.history.Logs.Chunk = b
l = &LogsMessage{Chunk: chunk}
}
db.logLock.Unlock()

21
dashboard/message.go
View file

@ -60,6 +60,27 @@ type NetworkMessage struct {
PeerBundles map[string]*PeerBundle `json:"peerBundles,omitempty"`
}
// getOrInitBundle returns the peer bundle belonging to the given IP, or
// initializes the bundle if it doesn't exist.
func (m *NetworkMessage) getOrInitBundle(ip string) *PeerBundle {
if _, ok := m.PeerBundles[ip]; !ok {
m.PeerBundles[ip] = &PeerBundle{
Peers: make(map[string]*Peer),
}
}
return m.PeerBundles[ip]
}
// getOrInitPeer returns the peer belonging to the given IP and node id, or
// initializes the peer if it doesn't exist.
func (m *NetworkMessage) getOrInitPeer(ip, id string) *Peer {
b := m.getOrInitBundle(ip)
if _, ok := b.Peers[id]; !ok {
b.Peers[id] = new(Peer)
}
return b.Peers[id]
}
// PeerBundle contains information about the peers pertaining to an IP address.
type PeerBundle struct {
Location *GeoLocation `json:"location,omitempty"` // geographical information based on IP

327
dashboard/peers.go
View file

@ -26,27 +26,6 @@ import (
const eventBufferLimit = 128 // Maximum number of buffered peer events for each event type
// getOrInitBundle returns the peer bundle belonging to the given IP, or
// initializes the bundle if it doesn't exist.
func getOrInitBundle(m *NetworkMessage, ip string) *PeerBundle {
if _, ok := m.PeerBundles[ip]; !ok {
m.PeerBundles[ip] = &PeerBundle{
Peers: make(map[string]*Peer),
}
}
return m.PeerBundles[ip]
}
// getOrInitPeer returns the peer belonging to the given IP and node id, or
// initializes the peer if it doesn't exist.
func getOrInitPeer(m *NetworkMessage, ip, id string) *Peer {
b := getOrInitBundle(m, ip)
if _, ok := b.Peers[id]; !ok {
b.Peers[id] = new(Peer)
}
return b.Peers[id]
}
// collectPeerData gathers data about the peers and sends it to the clients.
func (db *Dashboard) collectPeerData() {
defer db.wg.Done()
@ -60,94 +39,31 @@ func (db *Dashboard) collectPeerData() {
}
defer db.geodb.Close()
var (
quit = make(chan struct{})
// Channels used for avoiding the blocking of the event feeds.
connectCh = make(chan *p2p.PeerConnectEvent, eventBufferLimit)
handshakeCh = make(chan *p2p.PeerHandshakeEvent, eventBufferLimit)
disconnectCh = make(chan *p2p.PeerDisconnectEvent, eventBufferLimit)
readCh = make(chan *p2p.PeerReadEvent, eventBufferLimit)
writeCh = make(chan *p2p.PeerWriteEvent, eventBufferLimit)
)
go func() {
var (
// Peer event channels.
peerConnectEventCh = make(chan p2p.PeerConnectEvent, eventBufferLimit)
peerHandshakeEventCh = make(chan p2p.PeerHandshakeEvent, eventBufferLimit)
peerDisconnectEventCh = make(chan p2p.PeerDisconnectEvent, eventBufferLimit)
peerReadEventCh = make(chan p2p.PeerReadEvent, eventBufferLimit)
peerWriteEventCh = make(chan p2p.PeerWriteEvent, eventBufferLimit)
connectCh = make(chan p2p.PeerConnectEvent, eventBufferLimit)
handshakeCh = make(chan p2p.PeerHandshakeEvent, eventBufferLimit)
disconnectCh = make(chan p2p.PeerDisconnectEvent, eventBufferLimit)
//readCh = make(chan p2p.PeerReadEvent, eventBufferLimit)
//writeCh = make(chan p2p.PeerWriteEvent, eventBufferLimit)
// Subscribe to peer events.
subConnect = p2p.SubscribePeerConnectEvent(peerConnectEventCh)
subHandshake = p2p.SubscribePeerHandshakeEvent(peerHandshakeEventCh)
subDisconnect = p2p.SubscribePeerDisconnectEvent(peerDisconnectEventCh)
subRead = p2p.SubscribePeerReadEvent(peerReadEventCh)
subWrite = p2p.SubscribePeerWriteEvent(peerWriteEventCh)
subConnect = p2p.SubscribePeerConnectEvent(connectCh)
subHandshake = p2p.SubscribePeerHandshakeEvent(handshakeCh)
subDisconnect = p2p.SubscribePeerDisconnectEvent(disconnectCh)
//subRead = p2p.SubscribePeerReadEvent(readCh)
//subWrite = p2p.SubscribePeerWriteEvent(writeCh)
)
defer func() {
// Unsubscribe at the end.
subConnect.Unsubscribe()
subHandshake.Unsubscribe()
subDisconnect.Unsubscribe()
subRead.Unsubscribe()
subWrite.Unsubscribe()
//subRead.Unsubscribe()
//subWrite.Unsubscribe()
}()
// Waiting for peer events.
for {
select {
case event := <-peerConnectEventCh:
select {
case connectCh <- &event:
default:
log.Warn("Failed to handle peer connect event", "event", event)
}
case event := <-peerHandshakeEventCh:
select {
case handshakeCh <- &event:
default:
log.Warn("Failed to handle peer handshake event", "event", event)
}
case event := <-peerDisconnectEventCh:
select {
case disconnectCh <- &event:
default:
log.Warn("Failed to handle peer disconnect event", "event", event)
}
case event := <-peerReadEventCh:
select {
case readCh <- &event:
default:
log.Warn("Failed to handle peer read event", "event", event)
}
case event := <-peerWriteEventCh:
select {
case writeCh <- &event:
default:
log.Warn("Failed to handle peer write event", "event", event)
}
case err := <-subConnect.Err():
log.Warn("Peer connect subscription error", "err", err)
return
case err := <-subHandshake.Err():
log.Warn("Peer handshake subscription error", "err", err)
return
case err := <-subDisconnect.Err():
log.Warn("Peer disconnect subscription error", "err", err)
return
case err := <-subRead.Err():
log.Warn("Peer read subscription error", "err", err)
return
case err := <-subWrite.Err():
log.Warn("Peer write subscription error", "err", err)
return
case <-quit:
return
}
}
}()
go db.keepPeerHistoryClean(quit)
//go db.keepPeerHistoryClean(quit)
ticker := time.NewTicker(db.config.Refresh)
defer ticker.Stop()
@ -158,86 +74,86 @@ func (db *Dashboard) collectPeerData() {
}
for {
select {
case event := <-connectCh:
ip := event.IP.String()
p := getOrInitPeer(diff, ip, event.ID)
if diff.PeerBundles[ip].Location == nil {
db.peerLock.RLock()
lookup := db.networkHistory.PeerBundles[ip] == nil || db.networkHistory.PeerBundles[ip].Location == nil
db.peerLock.RUnlock()
if lookup {
location := db.geodb.Lookup(event.IP)
diff.PeerBundles[ip].Location = &GeoLocation{
Country: location.Country.Names.English,
City: location.City.Names.English,
Latitude: location.Location.Latitude,
Longitude: location.Location.Longitude,
}
}
}
if p.Connected == nil {
p.Connected = []time.Time{event.Connected}
} else {
p.Connected = append(p.Connected, event.Connected)
}
case event := <-handshakeCh:
ip := event.IP.String()
p := getOrInitPeer(diff, ip, event.DefaultID)
p.DefaultID = event.DefaultID
if p.Handshake == nil {
p.Handshake = []time.Time{event.Handshake}
} else {
p.Handshake = append(p.Handshake, event.Handshake)
}
delete(diff.PeerBundles[ip].Peers, event.DefaultID)
getOrInitPeer(diff, ip, event.ID)
diff.PeerBundles[ip].Peers[event.ID] = p // TODO (kurkomisi): Merge instead in order to keep the previous connection.
// Remove the peer from history in case the metering was before the handshake.
db.peerLock.RLock()
stored := db.networkHistory.PeerBundles[ip] != nil && db.networkHistory.PeerBundles[ip].Peers[event.DefaultID] != nil
db.peerLock.RUnlock()
if stored {
db.peerLock.Lock()
hp := getOrInitPeer(db.networkHistory, ip, event.DefaultID)
delete(db.networkHistory.PeerBundles[ip].Peers, event.DefaultID)
getOrInitPeer(db.networkHistory, ip, event.ID)
db.networkHistory.PeerBundles[ip].Peers[event.ID] = hp // TODO (kurkomisi): Merge.
db.peerLock.Unlock()
}
case event := <-disconnectCh:
p := getOrInitPeer(diff, event.IP.String(), event.ID)
if p.Disconnected == nil {
p.Disconnected = []time.Time{event.Disconnected}
} else {
p.Disconnected = append(p.Disconnected, event.Disconnected)
}
case event := <-readCh:
// Sum up the ingress between two updates.
p := getOrInitPeer(diff, event.IP.String(), event.ID)
if len(p.Ingress) <= 0 {
p.Ingress = ChartEntries{&ChartEntry{Value: float64(event.Ingress)}}
} else {
p.Ingress[0].Value += float64(event.Ingress)
}
case event := <-writeCh:
// Sum up the egress between two updates.
p := getOrInitPeer(diff, event.IP.String(), event.ID)
if len(p.Egress) <= 0 {
p.Egress = ChartEntries{&ChartEntry{Value: float64(event.Egress)}}
} else {
p.Egress[0].Value += float64(event.Egress)
}
//case event := <-connectCh:
// ip := event.IP.String()
// p := diff.getOrInitPeer(ip, event.ID)
// if diff.PeerBundles[ip].Location == nil {
// db.peerLock.RLock()
// lookup := db.history.Network.PeerBundles[ip] == nil || db.history.Network.PeerBundles[ip].Location == nil
// db.peerLock.RUnlock()
// if lookup {
// location := db.geodb.Lookup(event.IP)
// diff.PeerBundles[ip].Location = &GeoLocation{
// Country: location.Country.Names.English,
// City: location.City.Names.English,
// Latitude: location.Location.Latitude,
// Longitude: location.Location.Longitude,
// }
// }
// }
// if p.Connected == nil {
// p.Connected = []time.Time{event.Connected}
// } else {
// p.Connected = append(p.Connected, event.Connected)
// }
//case event := <-handshakeCh:
// ip := event.IP.String()
// p := diff.getOrInitPeer(ip, event.AutoID)
// p.DefaultID = event.AutoID
// if p.Handshake == nil {
// p.Handshake = []time.Time{event.Handshake}
// } else {
// p.Handshake = append(p.Handshake, event.Handshake)
// }
// delete(diff.PeerBundles[ip].Peers, event.AutoID)
// diff.getOrInitPeer(ip, event.ID)
// diff.PeerBundles[ip].Peers[event.ID] = p // TODO (kurkomisi): Merge instead in order to keep the previous connection.
// // Remove the peer from history in case the metering was before the handshake.
// db.peerLock.RLock()
// stored := db.history.Network.PeerBundles[ip] != nil && db.history.Network.PeerBundles[ip].Peers[event.AutoID] != nil
// db.peerLock.RUnlock()
// if stored {
// db.peerLock.Lock()
// hp := db.history.Network.getOrInitPeer(ip, event.AutoID)
// delete(db.history.Network.PeerBundles[ip].Peers, event.AutoID)
// db.history.Network.getOrInitPeer(ip, event.ID)
// db.history.Network.PeerBundles[ip].Peers[event.ID] = hp // TODO (kurkomisi): Merge.
// db.peerLock.Unlock()
// }
//case event := <-disconnectCh:
// p := diff.getOrInitPeer(event.IP.String(), event.ID)
// if p.Disconnected == nil {
// p.Disconnected = []time.Time{event.Disconnected}
// } else {
// p.Disconnected = append(p.Disconnected, event.Disconnected)
// }
//case event := <-readCh:
// // Sum up the ingress between two updates.
// p := diff.getOrInitPeer(event.IP.String(), event.ID)
// if len(p.Ingress) <= 0 {
// p.Ingress = ChartEntries{&ChartEntry{Value: float64(event.Ingress)}}
// } else {
// p.Ingress[0].Value += float64(event.Ingress)
// }
//case event := <-writeCh:
// // Sum up the egress between two updates.
// p := diff.getOrInitPeer(event.IP.String(), event.ID)
// if len(p.Egress) <= 0 {
// p.Egress = ChartEntries{&ChartEntry{Value: float64(event.Egress)}}
// } else {
// p.Egress[0].Value += float64(event.Egress)
// }
case <-ticker.C:
now := time.Now()
// Merge the diff with the history.
db.peerLock.Lock()
for ip, bundle := range diff.PeerBundles {
if bundle.Location != nil {
b := getOrInitBundle(db.networkHistory, ip)
b := db.history.Network.getOrInitBundle(ip)
b.Location = bundle.Location
}
for id, peer := range bundle.Peers {
peerHistory := getOrInitPeer(db.networkHistory, ip, id)
peerHistory := db.history.Network.getOrInitPeer(ip, id)
if peer.Connected != nil {
peerHistory.Connected = append(peerHistory.Connected, peer.Connected...)
}
@ -287,68 +203,25 @@ func (db *Dashboard) collectPeerData() {
}
delete(diff.PeerBundles, ip)
}
case err := <-subConnect.Err():
log.Warn("Peer connect subscription error", "err", err)
return
case err := <-subHandshake.Err():
log.Warn("Peer handshake subscription error", "err", err)
return
case err := <-subDisconnect.Err():
log.Warn("Peer disconnect subscription error", "err", err)
return
//case err := <-subRead.Err():
// log.Warn("Peer read subscription error", "err", err)
// return
//case err := <-subWrite.Err():
// log.Warn("Peer write subscription error", "err", err)
// return
case errc := <-db.quit:
close(quit)
errc <- nil
return
}
}
}
// keepPeerHistoryClean purges the stored peer metrics with a given rate in
// order to decrease the load. The inactive peers that disconnected before
// the calculated time will be deleted. If the total amount of peers exceeds
// the limit, the surplus will be chosen from the disconnected ones in the
// iteration order, and will be deleted as well.
func (db *Dashboard) keepPeerHistoryClean(quit chan struct{}) {
cleanRate := db.config.Refresh * peerTrafficSampleLimit
for {
select {
case <-time.After(cleanRate):
validAfter := time.Now().Add(-cleanRate)
db.peerLock.Lock()
for ip, bundle := range db.networkHistory.PeerBundles {
bundle.Location = nil
for id, peer := range bundle.Peers {
if len(peer.Disconnected) > 0 && peer.Disconnected[len(peer.Disconnected)-1].Before(validAfter) {
bundle.Peers[id] = nil
delete(bundle.Peers, id)
}
}
if len(bundle.Peers) <= 0 {
delete(db.networkHistory.PeerBundles, ip)
}
}
// TODO (kurkomisi): Check the limit during the insertion.
var lenCount int
for _, bundle := range db.networkHistory.PeerBundles {
lenCount += len(bundle.Peers)
}
if lenCount > peerLimit {
outerLoop:
for ip, bundle := range db.networkHistory.PeerBundles {
bundle.Location = nil
for id, peer := range bundle.Peers {
if peer.Disconnected != nil {
bundle.Peers[id] = nil
delete(bundle.Peers, id)
lenCount--
if lenCount <= peerLimit {
if len(bundle.Peers) <= 0 {
delete(bundle.Peers, ip)
}
break outerLoop
}
}
}
if len(bundle.Peers) <= 0 {
delete(db.networkHistory.PeerBundles, ip)
}
}
}
db.peerLock.Unlock()
case <-quit:
return
}
}
}

17
dashboard/system.go
View file

@ -128,14 +128,15 @@ func (db *Dashboard) collectSystemData() {
Value: float64(deltaDiskWrite) / frequency,
}
db.sysLock.Lock()
db.sysHistory.ActiveMemory = append(db.sysHistory.ActiveMemory[1:], activeMemory)
db.sysHistory.VirtualMemory = append(db.sysHistory.VirtualMemory[1:], virtualMemory)
db.sysHistory.NetworkIngress = append(db.sysHistory.NetworkIngress[1:], networkIngress)
db.sysHistory.NetworkEgress = append(db.sysHistory.NetworkEgress[1:], networkEgress)
db.sysHistory.ProcessCPU = append(db.sysHistory.ProcessCPU[1:], processCPU)
db.sysHistory.SystemCPU = append(db.sysHistory.SystemCPU[1:], systemCPU)
db.sysHistory.DiskRead = append(db.sysHistory.DiskRead[1:], diskRead)
db.sysHistory.DiskWrite = append(db.sysHistory.DiskWrite[1:], diskWrite)
sys := db.history.System
sys.ActiveMemory = append(sys.ActiveMemory[1:], activeMemory)
sys.VirtualMemory = append(sys.VirtualMemory[1:], virtualMemory)
sys.NetworkIngress = append(sys.NetworkIngress[1:], networkIngress)
sys.NetworkEgress = append(sys.NetworkEgress[1:], networkEgress)
sys.ProcessCPU = append(sys.ProcessCPU[1:], processCPU)
sys.SystemCPU = append(sys.SystemCPU[1:], systemCPU)
sys.DiskRead = append(sys.DiskRead[1:], diskRead)
sys.DiskWrite = append(sys.DiskWrite[1:], diskWrite)
db.sysLock.Unlock()
db.sendToAll(&Message{

167
p2p/metrics.go
View file

@ -20,6 +20,7 @@ package p2p
import (
"net"
"strings"
"fmt"
"sync"
@ -37,6 +38,11 @@ const (
MetricsInboundTraffic = "p2p/InboundTraffic" // Name for the registered inbound traffic meter
MetricsOutboundConnects = "p2p/OutboundConnects" // Name for the registered outbound connects meter
MetricsOutboundTraffic = "p2p/OutboundTraffic" // Name for the registered outbound traffic meter
MetricsRegistryIngressPrefix = MetricsInboundTraffic + "/"
MetricsRegistryEgressPrefix = MetricsOutboundTraffic + "/"
MeteredPeerLimit = 1024
)
var (
@ -45,9 +51,13 @@ var (
egressConnectMeter = metrics.NewRegisteredMeter(MetricsOutboundConnects, nil) // meter counting the egress connections
egressTrafficMeter = metrics.NewRegisteredMeter(MetricsOutboundTraffic, nil) // meter metering the cumulative egress traffic
PeerIngressRegistry = metrics.NewPrefixedChildRegistry(metrics.DefaultRegistry, MetricsRegistryIngressPrefix)
PeerEgressRegistry = metrics.NewPrefixedChildRegistry(metrics.DefaultRegistry, MetricsRegistryEgressPrefix)
metricsFeed = new(peerMetricsFeed) // Peer event feed for metrics
defaultMeteredPeerID uint64 // Used to create unique id for the metered connection before the handshake
meteredPeerAutoID uint64 // Used to create unique id for the metered connection before the handshake
meteredPeerCount uint64
)
// peerMetricsFeed delivers the peer metrics to the subscribed channels.
@ -59,43 +69,35 @@ type peerMetricsFeed struct {
write event.Feed // Event feed to notify the amount of written bytes of a peer
scope event.SubscriptionScope // Facility to unsubscribe all the subscriptions at once
quit chan chan error
}
// PeerConnectEvent contains information about the connection of a peer.
type PeerConnectEvent struct {
IP net.IP
ID string
Key string
Connected time.Time
}
// PeerHandshakeEvent contains information about the handshake with a peer.
type PeerHandshakeEvent struct {
IP net.IP
DefaultID string
ID string
AutoKey string
Key string
Ingress int64
Egress int64
Handshake time.Time
}
// PeerDisconnectEvent contains information about the disconnection of a peer.
type PeerDisconnectEvent struct {
IP net.IP
ID string
Key string
Ingress int64
Egress int64
Disconnected time.Time
}
// PeerReadEvent contains information about the read operation of a peer.
type PeerReadEvent struct {
IP net.IP
ID string
Ingress int
}
// PeerWriteEvent contains information about the write operation of a peer.
type PeerWriteEvent struct {
IP net.IP
ID string
Egress int
}
type PeerTrafficEvent map[string]int64
// SubscribePeerConnectEvent registers a subscription of PeerConnectEvent
func SubscribePeerConnectEvent(ch chan<- PeerConnectEvent) event.Subscription {
@ -113,26 +115,67 @@ func SubscribePeerDisconnectEvent(ch chan<- PeerDisconnectEvent) event.Subscript
}
// SubscribePeerReadEvent registers a subscription of PeerReadEvent
func SubscribePeerReadEvent(ch chan<- PeerReadEvent) event.Subscription {
func SubscribePeerReadEvent(ch chan<- PeerTrafficEvent) event.Subscription {
return metricsFeed.scope.Track(metricsFeed.read.Subscribe(ch))
}
// SubscribePeerWriteEvent registers a subscription of PeerWriteEvent
func SubscribePeerWriteEvent(ch chan<- PeerWriteEvent) event.Subscription {
func SubscribePeerWriteEvent(ch chan<- PeerTrafficEvent) event.Subscription {
return metricsFeed.scope.Track(metricsFeed.write.Subscribe(ch))
}
func startTrafficNotifier(refresh time.Duration) {
metricsFeed.quit = make(chan chan error)
ticker := time.NewTicker(refresh)
defer ticker.Stop()
for {
select {
case <-ticker.C:
// send read and write
ingressEvents, egressEvents := make(PeerTrafficEvent), make(PeerTrafficEvent)
PeerIngressRegistry.Each(func(name string, i interface{}) {
if m, ok := i.(metrics.Meter); ok {
ingressEvents[strings.TrimPrefix(name, MetricsRegistryIngressPrefix)] = m.Count()
}
})
PeerEgressRegistry.Each(func(name string, i interface{}) {
if m, ok := i.(metrics.Meter); ok {
egressEvents[strings.TrimPrefix(name, MetricsRegistryEgressPrefix)] = m.Count()
}
})
metricsFeed.read.Send(ingressEvents)
metricsFeed.write.Send(egressEvents)
//fmt.Println(ingressEvents)
//fmt.Println(egressEvents)
//fmt.Println()
case errc := <-metricsFeed.quit:
errc <- nil
return
}
}
}
// closeMetricsFeed closes all the tracked subscriptions.
func closeMetricsFeed() {
if metricsFeed.quit != nil {
errc := make(chan error)
metricsFeed.quit <- errc
<-errc
}
metricsFeed.scope.Close()
PeerIngressRegistry.UnregisterAll()
PeerEgressRegistry.UnregisterAll()
}
// meteredConn is a wrapper around a net.Conn that meters both the
// inbound and outbound network traffic.
type meteredConn struct {
net.Conn // Network connection to wrap with metering
ip net.IP // The IP address of the peer
id string // The node id of the peer
ip string // The IP address of the peer
key string
ingressMeter metrics.Meter
egressMeter metrics.Meter
lock sync.RWMutex // Lock protecting the metered connection's internals
}
@ -146,25 +189,31 @@ func newMeteredConn(conn net.Conn, ingress bool, ip net.IP) net.Conn {
return conn
}
if ip.IsUnspecified() {
log.Warn("peer IP is unspecified")
log.Warn("Peer IP is unspecified")
return conn
}
if atomic.LoadUint64(&meteredPeerCount) >= MeteredPeerLimit {
log.Warn("Metered peer count reached the limit")
return conn
}
atomic.AddUint64(&meteredPeerCount, 1)
// Otherwise bump the connection counters and wrap the connection
if ingress {
ingressConnectMeter.Mark(1)
} else {
egressConnectMeter.Mark(1)
}
id := fmt.Sprintf("peer_%d", atomic.AddUint64(&defaultMeteredPeerID, 1))
key := fmt.Sprintf("%s/%s", ip.String(), fmt.Sprintf("peer_%d", atomic.AddUint64(&meteredPeerAutoID, 1)))
metricsFeed.connect.Send(PeerConnectEvent{
IP: ip,
ID: id,
Key: key,
Connected: time.Now(),
})
return &meteredConn{
Conn: conn,
ip: ip,
id: id,
key: key,
ip: ip.String(),
ingressMeter: metrics.NewRegisteredMeter(key, PeerIngressRegistry),
egressMeter: metrics.NewRegisteredMeter(key, PeerEgressRegistry),
}
}
@ -173,14 +222,7 @@ func newMeteredConn(conn net.Conn, ingress bool, ip net.IP) net.Conn {
func (c *meteredConn) Read(b []byte) (n int, err error) {
n, err = c.Conn.Read(b)
ingressTrafficMeter.Mark(int64(n))
c.lock.RLock()
id := c.id
c.lock.RUnlock()
metricsFeed.read.Send(PeerReadEvent{
IP: c.ip,
ID: id,
Ingress: n,
})
c.ingressMeter.Mark(int64(n))
return n, err
}
@ -189,40 +231,55 @@ func (c *meteredConn) Read(b []byte) (n int, err error) {
func (c *meteredConn) Write(b []byte) (n int, err error) {
n, err = c.Conn.Write(b)
egressTrafficMeter.Mark(int64(n))
c.lock.RLock()
id := c.id
c.lock.RUnlock()
metricsFeed.write.Send(PeerWriteEvent{
IP: c.ip,
ID: id,
Egress: n,
})
c.egressMeter.Mark(int64(n))
return n, err
}
// Close closes the underlying connection.
func (c *meteredConn) Close() error {
// Decrement the metered peer count.
atomic.AddUint64(&meteredPeerCount, ^uint64(0))
c.ingressMeter.Stop()
c.egressMeter.Stop()
c.lock.RLock()
id := c.id
c.lock.RUnlock()
key := c.key
metricsFeed.disconnect.Send(PeerDisconnectEvent{
IP: c.ip,
ID: id,
Key: key,
Ingress: c.ingressMeter.Count(),
Egress: c.egressMeter.Count(),
Disconnected: time.Now(),
})
c.lock.RUnlock()
PeerIngressRegistry.Unregister(key)
PeerEgressRegistry.Unregister(key)
return c.Conn.Close()
}
// handshakeDone changes the default id to the peer's node id.
func (c *meteredConn) handshakeDone(id discover.NodeID) {
c.ingressMeter.Stop()
c.egressMeter.Stop()
c.lock.Lock()
defaultID := c.id
c.id = id.String()
autoKey := c.key
key := fmt.Sprintf("%s/%s", c.ip, id.String())
ingressMeter := metrics.NewRegisteredMeter(key, PeerIngressRegistry)
egressMeter := metrics.NewRegisteredMeter(key, PeerEgressRegistry)
ingressMeter.Mark(c.ingressMeter.Count())
egressMeter.Mark(c.egressMeter.Count())
PeerIngressRegistry.Unregister(c.key)
PeerEgressRegistry.Unregister(c.key)
c.key = key
c.ingressMeter = ingressMeter
c.egressMeter = egressMeter
c.lock.Unlock()
metricsFeed.handshake.Send(PeerHandshakeEvent{
IP: c.ip,
DefaultID: defaultID,
ID: id.String(),
AutoKey: autoKey,
Key: key,
//Ingress: nil,
//Egress: nil,
Handshake: time.Now(),
})
}

3
p2p/server.go
View file

@ -387,8 +387,8 @@ func (srv *Server) Stop() {
}
close(srv.quit)
srv.lock.Unlock()
srv.loopWG.Wait()
closeMetricsFeed()
srv.loopWG.Wait()
}
// sharedUDPConn implements a shared connection. Write sends messages to the underlying connection while read returns
@ -542,6 +542,7 @@ func (srv *Server) Start() (err error) {
srv.loopWG.Add(1)
go srv.run(dialer)
go startTrafficNotifier(2 * time.Second)
srv.running = true
return nil
}