p2p: meter peer traffic, emit metered peer events

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)
 }
 

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 (
+const (
-	ingressConnectMeter = metrics.NewRegisteredMeter("p2p/InboundConnects", nil)
+	MetricsInboundConnects  = "p2p/InboundConnects"  // Name for the registered inbound connects meter
-	ingressTrafficMeter = metrics.NewRegisteredMeter("p2p/InboundTraffic", nil)
+	MetricsInboundTraffic   = "p2p/InboundTraffic"   // Name for the registered inbound traffic meter
-	egressConnectMeter  = metrics.NewRegisteredMeter("p2p/OutboundConnects", nil)
+	MetricsOutboundConnects = "p2p/OutboundConnects" // Name for the registered outbound connects meter
-	egressTrafficMeter  = metrics.NewRegisteredMeter("p2p/OutboundTraffic", nil)
+	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
+// newMeteredConn creates a new metered connection, bumps the ingress or egress
-// egress connection meter. If the metrics system is disabled, this function
+// connection meter and also increases the metered peer count. If the metrics
-// returns the original object.
+// system is disabled, the IP address is unspecified or the metered peer count
-func newMeteredConn(conn net.Conn, ingress bool) net.Conn {
+// 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
+// Read delegates a network read to the underlying connection, bumping the common
-// traffic meter along the way.
+// 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
+// Write delegates a network write to the underlying connection, bumping the common
-// egress traffic meter along the way.
+// 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()
 }

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 {