From 9a4fb4a7ed83be10796954a56817f8bda06625ac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kurk=C3=B3=20Mih=C3=A1ly?= Date: Tue, 18 Sep 2018 19:17:16 +0300 Subject: [PATCH] p2p: meter peer traffic, emit metered peer events --- p2p/dial.go | 2 +- p2p/metrics.go | 177 ++++++++++++++++++++++++++++++++++++++++++++----- p2p/server.go | 11 ++- 3 files changed, 170 insertions(+), 20 deletions(-) diff --git a/p2p/dial.go b/p2p/dial.go index d228514fc6..075a0f9368 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 2d52fd1fd1..f78de71e78 100644 --- a/p2p/metrics.go +++ b/p2p/metrics.go @@ -19,53 +19,196 @@ package p2p import ( + "fmt" + "github.com/ethereum/go-ethereum/p2p/enode" "net" + "sync" + "sync/atomic" + "time" + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/metrics" ) -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) +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 + + MetricsRegistryIngressPrefix = MetricsInboundTraffic + "/" + MetricsRegistryEgressPrefix = MetricsOutboundTraffic + "/" + + MeteredPeerLimit = 1024 ) +var ( + 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, MetricsRegistryIngressPrefix) // Registry containing the peer ingress + PeerEgressRegistry = metrics.NewPrefixedChildRegistry(metrics.DefaultRegistry, MetricsRegistryEgressPrefix) // Registry containing the peer egress + + metricsFeed event.Feed // Event feed for peer metrics + meteredPeerCount uint64 // 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 in the moment of disconnection + Egress uint64 // Egress count in the moment of disconnection +} + +// SubscribeMeteredPeerEvent registers a subscription of MeteredPeerEvent +func SubscribeMeteredPeerEvent(ch chan<- MeteredPeerEvent) event.Subscription { + return metricsFeed.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 + 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, the IP address is unspecified or the metered peer count +// reached the limit, 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 + } + if atomic.LoadUint64(&meteredPeerCount) >= MeteredPeerLimit { + log.Warn("Metered peer count reached the limit") + return conn + } + // Increment the metered peer count + atomic.AddUint64(&meteredPeerCount, 1) + // 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.ingressMeter != nil { + 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.egressMeter != nil { + 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 NodeID, +// also emits connect event. +func (c *meteredConn) handshakeDone(id enode.ID) { + c.lock.Lock() + c.id = id.String() + key := fmt.Sprintf("%s/%s", c.ip, c.id) + c.ingressMeter = metrics.NewRegisteredMeter(key, PeerIngressRegistry) + c.egressMeter = metrics.NewRegisteredMeter(key, PeerEgressRegistry) + c.lock.Unlock() + + metricsFeed.Send(MeteredPeerEvent{ + Type: PeerConnected, + IP: c.ip, + ID: id.String(), + Elapsed: time.Now().Sub(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 { + // Decrement the metered peer count + atomic.AddUint64(&meteredPeerCount, ^uint64(0)) + + c.lock.RLock() + // If the peer disconnects before the handshake + if c.id == "" { + c.lock.RUnlock() + metricsFeed.Send(MeteredPeerEvent{ + Type: PeerHandshakeFailed, + IP: c.ip, + Elapsed: time.Now().Sub(c.connected), + }) + return c.Conn.Close() + } + id, ingress, egress := c.id, uint64(c.ingressMeter.Count()), uint64(c.egressMeter.Count()) + c.lock.RUnlock() + + // Unregister the peer from the traffic registries + key := fmt.Sprintf("%s/%s", c.ip, id) + PeerIngressRegistry.Unregister(key) + PeerEgressRegistry.Unregister(key) + + metricsFeed.Send(MeteredPeerEvent{ + Type: PeerDisconnected, + IP: c.ip, + ID: id, + Ingress: ingress, + Egress: egress, + }) + return c.Conn.Close() } diff --git a/p2p/server.go b/p2p/server.go index 6482c04019..91ee78fe95 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -84,7 +84,7 @@ type Config struct { // Disabling is useful for protocol debugging (manual topology). NoDiscovery bool - // DiscoveryV5 specifies whether the new topic-discovery based V5 discovery + // DiscoveryV5 specifies whether the the new topic-discovery based V5 discovery // protocol should be started or not. DiscoveryV5 bool `toml:",omitempty"` @@ -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 {