p2p: Add metrics for tracking p2p modules

metrics/gauge.go

@@ -6,6 +6,8 @@ import "sync/atomic"
 type Gauge interface {
 	Snapshot() Gauge
 	Update(int64)
+	Dec(int64)
+	Inc(int64)
 	Value() int64
 }
 
@@ -65,6 +67,16 @@ func (GaugeSnapshot) Update(int64) {
 	panic("Update called on a GaugeSnapshot")
 }
 
+// Dec panics.
+func (GaugeSnapshot) Dec(int64) {
+	panic("Dec called on a GaugeSnapshot")
+}
+
+// Inc panics.
+func (GaugeSnapshot) Inc(int64) {
+	panic("Inc called on a GaugeSnapshot")
+}
+
 // Value returns the value at the time the snapshot was taken.
 func (g GaugeSnapshot) Value() int64 { return int64(g) }
 
@@ -77,6 +89,12 @@ func (NilGauge) Snapshot() Gauge { return NilGauge{} }
 // Update is a no-op.
 func (NilGauge) Update(v int64) {}
 
+// Dec is a no-op.
+func (NilGauge) Dec(i int64) {}
+
+// Inc is a no-op.
+func (NilGauge) Inc(i int64) {}
+
 // Value is a no-op.
 func (NilGauge) Value() int64 { return 0 }
 
@@ -96,6 +114,16 @@ func (g *StandardGauge) Update(v int64) {
 	atomic.StoreInt64(&g.value, v)
 }
 
+// Dec decrements the gauge's current value by the given amount.
+func (g *StandardGauge) Dec(i int64) {
+	atomic.AddInt64(&g.value, -i)
+}
+
+// Inc increments the gauge's current value by the given amount.
+func (g *StandardGauge) Inc(i int64) {
+	atomic.AddInt64(&g.value, i)
+}
+
 // Value returns the gauge's current value.
 func (g *StandardGauge) Value() int64 {
 	return atomic.LoadInt64(&g.value)
@@ -118,3 +146,13 @@ func (g FunctionalGauge) Snapshot() Gauge { return GaugeSnapshot(g.Value()) }
 func (FunctionalGauge) Update(int64) {
 	panic("Update called on a FunctionalGauge")
 }
+
+// Dec panics.
+func (FunctionalGauge) Dec(int64) {
+	panic("Dec called on a FunctionalGauge")
+}
+
+// Inc panics.
+func (FunctionalGauge) Inc(int64) {
+	panic("Inc called on a FunctionalGauge")
+}

p2p/dial.go

@@ -146,7 +146,7 @@ func newDialState(static []*discover.Node, bootnodes []*discover.Node, ntab disc
 }
 
 func (s *dialstate) addStatic(n *discover.Node) {
-	// This overwites the task instead of updating an existing
+	// This overwrites the task instead of updating an existing
 	// entry, giving users the opportunity to force a resolve operation.
 	s.static[n.ID] = &dialTask{flags: staticDialedConn, dest: n}
 }

p2p/message.go

@@ -39,9 +39,13 @@ import (
 // separate Msg with a bytes.Reader as Payload for each send.
 type Msg struct {
 	Code       uint64
-	Size       uint32 // size of the paylod
+	Size       uint32 // Size of the raw payload
 	Payload    io.Reader
 	ReceivedAt time.Time
+
+	meterCap  Cap    // Protocol name and version for egress metering
+	meterCode uint64 // Message within protocol for egress metering
+	meterSize uint32 // Compressed message size for ingress metering
 }
 
 // Decode parses the RLP content of a message into
@@ -100,12 +104,11 @@ func Send(w MsgWriter, msgcode uint64, data interface{}) error {
 // SendItems writes an RLP with the given code and data elements.
 // For a call such as:
 //
-//    SendItems(w, code, e1, e2, e3)
+//	SendItems(w, code, e1, e2, e3)
 //
 // the message payload will be an RLP list containing the items:
 //
-//    [e1, e2, e3]
-//
+//	[e1, e2, e3]
 func SendItems(w MsgWriter, msgcode uint64, elems ...interface{}) error {
 	return Send(w, msgcode, elems)
 }

p2p/metrics.go

@@ -24,11 +24,20 @@ import (
 	"github.com/tomochain/tomochain/metrics"
 )
 
+const (
+	// ingressMeterName is the prefix of the per-packet inbound metrics.
+	ingressMeterName = "p2p/ingress"
+
+	// egressMeterName is the prefix of the per-packet outbound metrics.
+	egressMeterName = "p2p/egress"
+)
+
 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("p2p/serves", nil)
+	ingressTrafficMeter = metrics.NewRegisteredMeter(ingressMeterName, nil)
+	egressConnectMeter  = metrics.NewRegisteredMeter("p2p/dials", nil)
+	egressTrafficMeter  = metrics.NewRegisteredMeter(egressMeterName, nil)
+	activePeerGauge     = metrics.NewRegisteredGauge("p2p/peers", nil)
 )
 
 // meteredConn is a wrapper around a network TCP connection that meters both the
@@ -51,6 +60,7 @@ func newMeteredConn(conn net.Conn, ingress bool) net.Conn {
 	} else {
 		egressConnectMeter.Mark(1)
 	}
+	activePeerGauge.Inc(1)
 	return &meteredConn{conn.(*net.TCPConn)}
 }
 
@@ -69,3 +79,13 @@ func (c *meteredConn) Write(b []byte) (n int, err error) {
 	egressTrafficMeter.Mark(int64(n))
 	return
 }
+
+// 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.TCPConn.Close()
+	if err == nil {
+		activePeerGauge.Dec(1)
+	}
+	return err
+}

p2p/peer.go

@@ -27,6 +27,7 @@ import (
 	"github.com/tomochain/tomochain/common/mclock"
 	"github.com/tomochain/tomochain/event"
 	"github.com/tomochain/tomochain/log"
+	"github.com/tomochain/tomochain/metrics"
 	"github.com/tomochain/tomochain/p2p/discover"
 	"github.com/tomochain/tomochain/rlp"
 )
@@ -289,6 +290,11 @@ func (p *Peer) handle(msg Msg) error {
 		if err != nil {
 			return fmt.Errorf("msg code out of range: %v", msg.Code)
 		}
+		if metrics.Enabled {
+			m := fmt.Sprintf("%s/%s/%d/%#02x", ingressMeterName, proto.Name, proto.Version, msg.Code-proto.offset)
+			metrics.GetOrRegisterMeter(m, nil).Mark(int64(msg.meterSize))
+			metrics.GetOrRegisterMeter(m+"/packets", nil).Mark(1)
+		}
 		select {
 		case proto.in <- msg:
 			return nil
@@ -388,6 +394,9 @@ func (rw *protoRW) WriteMsg(msg Msg) (err error) {
 	if msg.Code >= rw.Length {
 		return newPeerError(errInvalidMsgCode, "not handled")
 	}
+	msg.meterCap = rw.cap()
+	msg.meterCode = msg.Code
+
 	msg.Code += rw.offset
 	select {
 	case <-rw.wstart:

p2p/rlpx.go

@@ -40,6 +40,7 @@ import (
 	"github.com/tomochain/tomochain/crypto/ecies"
 	"github.com/tomochain/tomochain/crypto/secp256k1"
 	"github.com/tomochain/tomochain/crypto/sha3"
+	"github.com/tomochain/tomochain/metrics"
 	"github.com/tomochain/tomochain/p2p/discover"
 	"github.com/tomochain/tomochain/rlp"
 )
@@ -612,6 +613,10 @@ func (rw *rlpxFrameRW) WriteMsg(msg Msg) error {
 		msg.Payload = bytes.NewReader(payload)
 		msg.Size = uint32(len(payload))
 	}
+	msg.meterSize = msg.Size
+	if metrics.Enabled && msg.meterCap.Name != "" { // don't meter non-subprotocol messages
+		metrics.GetOrRegisterMeter(fmt.Sprintf("%s/%s/%d/%#02x", egressMeterName, msg.meterCap.Name, msg.meterCap.Version, msg.meterCode), nil).Mark(int64(msg.meterSize))
+	}
 	// write header
 	headbuf := make([]byte, 32)
 	fsize := uint32(len(ptype)) + msg.Size