diff options
| -rw-r--r-- | p2p/dial.go | 2 | ||||
| -rw-r--r-- | p2p/metrics.go | 194 | ||||
| -rw-r--r-- | p2p/server.go | 9 |
3 files changed, 187 insertions, 18 deletions
diff --git a/p2p/dial.go b/p2p/dial.go index d228514fc..075a0f936 100644 --- a/p2p/dial.go +++ b/p2p/dial.go @@ -350,7 +350,7 @@ func (t *dialTask) dial(srv *Server, dest *enode.Node) error { if err != nil { return &dialError{err} } - mfd := newMeteredConn(fd, false) + mfd := newMeteredConn(fd, false, dest.IP()) return srv.SetupConn(mfd, t.flags, dest) } diff --git a/p2p/metrics.go b/p2p/metrics.go index 2d52fd1fd..6a7c0bad3 100644 --- a/p2p/metrics.go +++ b/p2p/metrics.go @@ -19,53 +19,215 @@ package p2p import ( + "fmt" "net" + "sync" + "sync/atomic" + "time" + "github.com/ethereum/go-ethereum/p2p/enode" + + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/metrics" ) +const ( + MetricsInboundConnects = "p2p/InboundConnects" // Name for the registered inbound connects meter + 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 + + MeteredPeerLimit = 1024 // This amount of peers are individually metered +) + var ( - ingressConnectMeter = metrics.NewRegisteredMeter("p2p/InboundConnects", nil) - ingressTrafficMeter = metrics.NewRegisteredMeter("p2p/InboundTraffic", nil) - egressConnectMeter = metrics.NewRegisteredMeter("p2p/OutboundConnects", nil) - egressTrafficMeter = metrics.NewRegisteredMeter("p2p/OutboundTraffic", nil) + ingressConnectMeter = metrics.NewRegisteredMeter(MetricsInboundConnects, nil) // Meter counting the ingress connections + ingressTrafficMeter = metrics.NewRegisteredMeter(MetricsInboundTraffic, nil) // Meter metering the cumulative ingress traffic + 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, MetricsInboundTraffic+"/") // Registry containing the peer ingress + PeerEgressRegistry = metrics.NewPrefixedChildRegistry(metrics.DefaultRegistry, MetricsOutboundTraffic+"/") // Registry containing the peer egress + + meteredPeerFeed event.Feed // Event feed for peer metrics + meteredPeerCount int32 // Actually stored peer connection count ) +// MeteredPeerEventType is the type of peer events emitted by a metered connection. +type MeteredPeerEventType int + +const ( + // PeerConnected is the type of event emitted when a peer successfully + // made the handshake. + PeerConnected MeteredPeerEventType = iota + + // PeerDisconnected is the type of event emitted when a peer disconnects. + PeerDisconnected + + // PeerHandshakeFailed is the type of event emitted when a peer fails to + // make the handshake or disconnects before the handshake. + PeerHandshakeFailed +) + +// MeteredPeerEvent is an event emitted when peers connect or disconnect. +type MeteredPeerEvent struct { + Type MeteredPeerEventType // Type of peer event + IP net.IP // IP address of the peer + ID string // NodeID of the peer + Elapsed time.Duration // Time elapsed between the connection and the handshake/disconnection + Ingress uint64 // Ingress count at the moment of the event + Egress uint64 // Egress count at the moment of the event +} + +// SubscribeMeteredPeerEvent registers a subscription for peer life-cycle events +// if metrics collection is enabled. +func SubscribeMeteredPeerEvent(ch chan<- MeteredPeerEvent) event.Subscription { + return meteredPeerFeed.Subscribe(ch) +} + // 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 + + connected time.Time // Connection time of the peer + ip net.IP // IP address of the peer + id string // NodeID of the peer + + // trafficMetered denotes if the peer is registered in the traffic registries. + // Its value is true if the metered peer count doesn't reach the limit in the + // moment of the peer's connection. + trafficMetered bool + ingressMeter metrics.Meter // Meter for the read bytes of the peer + egressMeter metrics.Meter // Meter for the written bytes of the peer + + lock sync.RWMutex // Lock protecting the metered connection's internals } -// newMeteredConn creates a new metered connection, also bumping the ingress or -// egress connection meter. If the metrics system is disabled, this function -// returns the original object. -func newMeteredConn(conn net.Conn, ingress bool) net.Conn { +// newMeteredConn creates a new metered connection, bumps the ingress or egress +// connection meter and also increases the metered peer count. If the metrics +// system is disabled or the IP address is unspecified, this function returns +// the original object. +func newMeteredConn(conn net.Conn, ingress bool, ip net.IP) net.Conn { // Short circuit if metrics are disabled if !metrics.Enabled { return conn } - // Otherwise bump the connection counters and wrap the connection + if ip.IsUnspecified() { + log.Warn("Peer IP is unspecified") + return conn + } + // Bump the connection counters and wrap the connection if ingress { ingressConnectMeter.Mark(1) } else { egressConnectMeter.Mark(1) } - return &meteredConn{Conn: conn} + return &meteredConn{ + Conn: conn, + ip: ip, + connected: time.Now(), + } } -// Read delegates a network read to the underlying connection, bumping the ingress -// traffic meter along the way. +// Read delegates a network read to the underlying connection, bumping the common +// and the peer ingress traffic meters along the way. func (c *meteredConn) Read(b []byte) (n int, err error) { n, err = c.Conn.Read(b) ingressTrafficMeter.Mark(int64(n)) - return + c.lock.RLock() + if c.trafficMetered { + c.ingressMeter.Mark(int64(n)) + } + c.lock.RUnlock() + return n, err } -// Write delegates a network write to the underlying connection, bumping the -// egress traffic meter along the way. +// Write delegates a network write to the underlying connection, bumping the common +// and the peer egress traffic meters along the way. func (c *meteredConn) Write(b []byte) (n int, err error) { n, err = c.Conn.Write(b) egressTrafficMeter.Mark(int64(n)) - return + c.lock.RLock() + if c.trafficMetered { + c.egressMeter.Mark(int64(n)) + } + c.lock.RUnlock() + return n, err +} + +// handshakeDone is called when a peer handshake is done. Registers the peer to +// the ingress and the egress traffic registries using the peer's IP and node ID, +// also emits connect event. +func (c *meteredConn) handshakeDone(nodeID enode.ID) { + id := nodeID.String() + if atomic.AddInt32(&meteredPeerCount, 1) >= MeteredPeerLimit { + // Don't register the peer in the traffic registries. + atomic.AddInt32(&meteredPeerCount, -1) + c.lock.Lock() + c.id, c.trafficMetered = id, false + c.lock.Unlock() + log.Warn("Metered peer count reached the limit") + } else { + key := fmt.Sprintf("%s/%s", c.ip, id) + c.lock.Lock() + c.id, c.trafficMetered = id, true + c.ingressMeter = metrics.NewRegisteredMeter(key, PeerIngressRegistry) + c.egressMeter = metrics.NewRegisteredMeter(key, PeerEgressRegistry) + c.lock.Unlock() + } + meteredPeerFeed.Send(MeteredPeerEvent{ + Type: PeerConnected, + IP: c.ip, + ID: id, + Elapsed: time.Since(c.connected), + }) +} + +// Close delegates a close operation to the underlying connection, unregisters +// the peer from the traffic registries and emits close event. +func (c *meteredConn) Close() error { + err := c.Conn.Close() + c.lock.RLock() + if c.id == "" { + // If the peer disconnects before the handshake. + c.lock.RUnlock() + meteredPeerFeed.Send(MeteredPeerEvent{ + Type: PeerHandshakeFailed, + IP: c.ip, + Elapsed: time.Since(c.connected), + }) + return err + } + id := c.id + if !c.trafficMetered { + // If the peer isn't registered in the traffic registries. + c.lock.RUnlock() + meteredPeerFeed.Send(MeteredPeerEvent{ + Type: PeerDisconnected, + IP: c.ip, + ID: id, + }) + return err + } + ingress, egress := uint64(c.ingressMeter.Count()), uint64(c.egressMeter.Count()) + c.lock.RUnlock() + + // Decrement the metered peer count + atomic.AddInt32(&meteredPeerCount, -1) + + // Unregister the peer from the traffic registries + key := fmt.Sprintf("%s/%s", c.ip, id) + PeerIngressRegistry.Unregister(key) + PeerEgressRegistry.Unregister(key) + + meteredPeerFeed.Send(MeteredPeerEvent{ + Type: PeerDisconnected, + IP: c.ip, + ID: id, + Ingress: ingress, + Egress: egress, + }) + return err } diff --git a/p2p/server.go b/p2p/server.go index 6482c0401..38a881f7b 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -864,7 +864,11 @@ func (srv *Server) listenLoop() { } } - fd = newMeteredConn(fd, true) + var ip net.IP + if tcp, ok := fd.RemoteAddr().(*net.TCPAddr); ok { + ip = tcp.IP + } + fd = newMeteredConn(fd, true, ip) srv.log.Trace("Accepted connection", "addr", fd.RemoteAddr()) go func() { srv.SetupConn(fd, inboundConn, nil) @@ -917,6 +921,9 @@ func (srv *Server) setupConn(c *conn, flags connFlag, dialDest *enode.Node) erro } else { c.node = nodeFromConn(remotePubkey, c.fd) } + if conn, ok := c.fd.(*meteredConn); ok { + conn.handshakeDone(c.node.ID()) + } clog := srv.log.New("id", c.node.ID(), "addr", c.fd.RemoteAddr(), "conn", c.flags) err = srv.checkpoint(c, srv.posthandshake) if err != nil { |