diff --git a/metrics/gauge.go b/metrics/gauge.go index 0fbfdb860..4b113b4d3 100644 --- a/metrics/gauge.go +++ b/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") +} diff --git a/p2p/dial.go b/p2p/dial.go index 454d2198c..d4623b303 100644 --- a/p2p/dial.go +++ b/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} } diff --git a/p2p/message.go b/p2p/message.go index c92b31581..ac89b72ae 100644 --- a/p2p/message.go +++ b/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) } diff --git a/p2p/metrics.go b/p2p/metrics.go index 905d5f4c1..30dd69cb3 100644 --- a/p2p/metrics.go +++ b/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 +} diff --git a/p2p/peer.go b/p2p/peer.go index 1852d59bc..77bcfefe2 100644 --- a/p2p/peer.go +++ b/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: diff --git a/p2p/rlpx.go b/p2p/rlpx.go index 2cc4d42d3..6055a63de 100644 --- a/p2p/rlpx.go +++ b/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