diff --git a/dashboard/assets/components/SideBar.jsx b/dashboard/assets/components/SideBar.jsx index 1c5df6da27..6082152f1d 100644 --- a/dashboard/assets/components/SideBar.jsx +++ b/dashboard/assets/components/SideBar.jsx @@ -51,6 +51,7 @@ const themeStyles = theme => ({ }, icon: { fontSize: theme.spacing.unit * 3, + overflow: 'unset', }, }); diff --git a/dashboard/assets/webpack.config.js b/dashboard/assets/webpack.config.js deleted file mode 100644 index 71beeeb41f..0000000000 --- a/dashboard/assets/webpack.config.js +++ /dev/null @@ -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 . - -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, - }, -}; diff --git a/dashboard/dashboard.go b/dashboard/dashboard.go index f33b233f36..10beceaa25 100644 --- a/dashboard/dashboard.go +++ b/dashboard/dashboard.go @@ -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,19 +102,19 @@ 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{ - ActiveMemory: emptyChartEntries(now, activeMemorySampleLimit, config.Refresh), - VirtualMemory: emptyChartEntries(now, virtualMemorySampleLimit, config.Refresh), - NetworkIngress: emptyChartEntries(now, networkIngressSampleLimit, config.Refresh), - NetworkEgress: emptyChartEntries(now, networkEgressSampleLimit, config.Refresh), - ProcessCPU: emptyChartEntries(now, processCPUSampleLimit, config.Refresh), - SystemCPU: emptyChartEntries(now, systemCPUSampleLimit, config.Refresh), - DiskRead: emptyChartEntries(now, diskReadSampleLimit, config.Refresh), - DiskWrite: emptyChartEntries(now, diskWriteSampleLimit, config.Refresh), - }, - networkHistory: &NetworkMessage{ - PeerBundles: make(map[string]*PeerBundle), + System: &SystemMessage{ + ActiveMemory: emptyChartEntries(now, activeMemorySampleLimit, config.Refresh), + VirtualMemory: emptyChartEntries(now, virtualMemorySampleLimit, config.Refresh), + NetworkIngress: emptyChartEntries(now, networkIngressSampleLimit, config.Refresh), + NetworkEgress: emptyChartEntries(now, networkEgressSampleLimit, config.Refresh), + ProcessCPU: emptyChartEntries(now, processCPUSampleLimit, config.Refresh), + SystemCPU: emptyChartEntries(now, systemCPUSampleLimit, config.Refresh), + DiskRead: emptyChartEntries(now, diskReadSampleLimit, config.Refresh), + DiskWrite: emptyChartEntries(now, diskWriteSampleLimit, config.Refresh), + }, + 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. diff --git a/dashboard/log.go b/dashboard/log.go index be245d5424..be7732695c 100644 --- a/dashboard/log.go +++ b/dashboard/log.go @@ -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() diff --git a/dashboard/message.go b/dashboard/message.go index 4e047025b1..dfe949bcce 100644 --- a/dashboard/message.go +++ b/dashboard/message.go @@ -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 diff --git a/dashboard/peers.go b/dashboard/peers.go index fb7445426e..eb64bb4ffc 100644 --- a/dashboard/peers.go +++ b/dashboard/peers.go @@ -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() @@ -61,93 +40,30 @@ func (db *Dashboard) collectPeerData() { defer db.geodb.Close() var ( - quit = make(chan struct{}) + // Peer event channels. + 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) - // 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) + // Subscribe to peer events. + subConnect = p2p.SubscribePeerConnectEvent(connectCh) + subHandshake = p2p.SubscribePeerHandshakeEvent(handshakeCh) + subDisconnect = p2p.SubscribePeerDisconnectEvent(disconnectCh) + //subRead = p2p.SubscribePeerReadEvent(readCh) + //subWrite = p2p.SubscribePeerWriteEvent(writeCh) ) - 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) - - // Subscribe to peer events. - subConnect = p2p.SubscribePeerConnectEvent(peerConnectEventCh) - subHandshake = p2p.SubscribePeerHandshakeEvent(peerHandshakeEventCh) - subDisconnect = p2p.SubscribePeerDisconnectEvent(peerDisconnectEventCh) - subRead = p2p.SubscribePeerReadEvent(peerReadEventCh) - subWrite = p2p.SubscribePeerWriteEvent(peerWriteEventCh) - ) - defer func() { - // Unsubscribe at the end. - subConnect.Unsubscribe() - subHandshake.Unsubscribe() - subDisconnect.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 - } - } + defer func() { + // Unsubscribe at the end. + subConnect.Unsubscribe() + subHandshake.Unsubscribe() + subDisconnect.Unsubscribe() + //subRead.Unsubscribe() + //subWrite.Unsubscribe() }() - 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 - } - } -} diff --git a/dashboard/system.go b/dashboard/system.go index 4842a6d841..ca9e6fc71b 100644 --- a/dashboard/system.go +++ b/dashboard/system.go @@ -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{ diff --git a/p2p/metrics.go b/p2p/metrics.go index 134f689bdf..9031201931 100644 --- a/p2p/metrics.go +++ b/p2p/metrics.go @@ -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, + Conn: conn, + 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(), }) } diff --git a/p2p/server.go b/p2p/server.go index c1a1ac9952..a9bd9c7c54 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -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 }