feat: update node metric to get BytesSend and BytesReceived

Author: soonkuk

isaac/network/node_metric.go

@@ -2,13 +2,15 @@ package isaacnetwork
 
 import (
 	"math"
+	"reflect"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/ProtoconNet/mitum2/network/quicstream"
 	"github.com/ProtoconNet/mitum2/util"
 	"github.com/ProtoconNet/mitum2/util/hint"
+	"github.com/ProtoconNet/mitum2/util/localtime"
 )
 
 var NodeMetricsHint = hint.MustNewHint("node-metrics-v0.0.1")
@@ -30,22 +32,64 @@ type NodeMetrics struct {
 
 func (m NodeMetrics) MarshalJSON() ([]byte, error) {
 	type alias struct {
-		Hint       hint.Type                  `json:"hint"`
-		Timestamp  time.Time                  `json:"timestamp"`
-		Uptime     string                     `json:"uptime"`
+		hint.BaseHinter
+		Timestamp  localtime.Time             `json:"timestamp"`
+		Uptime     util.ReadableDuration      `json:"uptime"`
 		Cumulative CumulativeMetrics          `json:"cumulative"`
 		Intervals  map[string]IntervalMetrics `json:"intervals"`
 	}
 
 	return util.MarshalJSON(alias{
-		Hint:       m.Hint().Type(),
-		Timestamp:  m.Timestamp,
-		Uptime:     m.Uptime.String(),
+		BaseHinter: m.BaseHinter,
+		Timestamp:  localtime.New(m.Timestamp),
+		Uptime:     util.ReadableDuration(m.Uptime),
 		Cumulative: m.Cumulative,
 		Intervals:  m.Intervals,
 	})
 }
 
+func (m *NodeMetrics) UnmarshalJSON(b []byte) error {
+	e := util.StringError("unmarshal NodeMetrics")
+
+	type alias struct {
+		hint.BaseHinter
+		Timestamp  time.Time                  `json:"timestamp"`
+		Uptime     *util.ReadableDuration     `json:"uptime"`
+		Cumulative CumulativeMetrics          `json:"cumulative"`
+		Intervals  map[string]IntervalMetrics `json:"intervals"`
+	}
+
+	var u alias
+
+	if err := util.UnmarshalJSON(b, &u); err != nil {
+		return e.Wrap(err)
+	}
+
+	m.BaseHinter = u.BaseHinter
+	m.Timestamp = u.Timestamp
+	m.Cumulative = u.Cumulative
+	m.Intervals = u.Intervals
+
+	durArgs := [][2]interface{}{
+		{u.Uptime, &m.Uptime},
+	}
+
+	for i := range durArgs {
+		v := durArgs[i][0].(*util.ReadableDuration) //nolint:forcetypeassert //...
+		t := durArgs[i][1].(*time.Duration)         //nolint:forcetypeassert //...
+
+		if reflect.ValueOf(v).IsZero() {
+			continue
+		}
+
+		if err := util.SetInterfaceValue(time.Duration(*v), t); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 type CumulativeMetrics struct {
 	QuicBytesSent          uint64 `json:"quic_bytes_sent"`
 	QuicBytesReceived      uint64 `json:"quic_bytes_received"`

isaac/network/quicstream_handlers.go

@@ -3,7 +3,6 @@ package isaacnetwork
 import (
 	"bytes"
 	"context"
-	"fmt"
 	"io"
 	"net"
 	"net/url"
@@ -542,7 +541,7 @@ func QuicstreamHandlerNodeInfo(
 		broker *quicstreamheader.HandlerBroker, _ NodeInfoRequestHeader,
 	) (context.Context, error) {
 		e := util.StringError("handle node info")
-		fmt.Println("QuicstreamHandlerNodeInfo QuicstreamHandlerNodeInfo")
+
 		b, err, _ := util.SingleflightDo[[]byte](&sg, HandlerNameNodeInfo.String(), func() ([]byte, error) {
 			return getNodeInfo()
 		})

launch/hinters.go

@@ -73,6 +73,7 @@ var Hinters = []encoder.DecodeDetail{
 	{Hint: isaacnetwork.NodeChallengeRequestHeaderHint, Instance: isaacnetwork.NodeChallengeRequestHeader{}},
 	{Hint: isaacnetwork.NodeInfoHint, Instance: isaacnetwork.NodeInfo{}},
 	{Hint: isaacnetwork.NodeInfoRequestHeaderHint, Instance: isaacnetwork.NodeInfoRequestHeader{}},
+	{Hint: isaacnetwork.NodeMetricsHint, Instance: isaacnetwork.NodeMetrics{}},
 	{Hint: isaacnetwork.NodeMetricsRequestHeaderHint, Instance: isaacnetwork.NodeMetricsRequestHeader{}},
 	{Hint: isaacnetwork.OperationRequestHeaderHint, Instance: isaacnetwork.OperationRequestHeader{}},
 	{Hint: isaacnetwork.ProposalRequestHeaderHint, Instance: isaacnetwork.ProposalRequestHeader{}},

network/quicstream/connection.go

@@ -57,13 +57,15 @@ func (c *Connection) Stream(ctx context.Context, f StreamFunc) error {
 		defer collector.RecordQuicStreamClosed()
 	}
 
+	r, w := wrapMetricsIO(ctx, stream, stream)
+
 	defer func() {
 		stream.CancelRead(0)
 		_ = stream.Close()
 	}()
 
 	return util.AwareContext(ctx, func(ctx context.Context) error {
-		return f(ctx, stream, stream)
+		return f(ctx, r, w)
 	})
 }
 
@@ -77,7 +79,9 @@ func (c *Connection) OpenStream(ctx context.Context) (io.Reader, io.WriteCloser,
 		collector.RecordQuicStreamOpened()
 	}
 
-	return stream, stream, func() error {
+	r, w := wrapMetricsIO(ctx, stream, stream)
+
+	return r, w, func() error {
 		if collector := GetMetricsCollector(ctx); collector != nil {
 			collector.RecordQuicStreamClosed()
 		}

network/quicstream/metrics.go

@@ -2,6 +2,7 @@ package quicstream
 
 import (
 	"context"
+	"io"
 
 	"github.com/ProtoconNet/mitum2/util"
 )
@@ -47,3 +48,47 @@ func GetMetricsCollector(ctx context.Context) MetricsCollector {
 
 	return nil
 }
+
+func wrapMetricsIO(ctx context.Context, reader io.Reader, writer io.WriteCloser) (io.Reader, io.WriteCloser) {
+	collector := GetMetricsCollector(ctx)
+
+	switch {
+	case collector == nil:
+		return reader, writer
+	case reader == nil:
+		return reader, &metricsWriteCloser{WriteCloser: writer, collector: collector}
+	case writer == nil:
+		return &metricsReader{reader: reader, collector: collector}, writer
+	default:
+		return &metricsReader{reader: reader, collector: collector},
+			&metricsWriteCloser{WriteCloser: writer, collector: collector}
+	}
+}
+
+type metricsReader struct {
+	reader    io.Reader
+	collector MetricsCollector
+}
+
+func (r *metricsReader) Read(p []byte) (int, error) {
+	n, err := r.reader.Read(p)
+	if n > 0 && r.collector != nil {
+		r.collector.RecordQuicBytesReceived(uint64(n))
+	}
+
+	return n, err
+}
+
+type metricsWriteCloser struct {
+	io.WriteCloser
+	collector MetricsCollector
+}
+
+func (w *metricsWriteCloser) Write(p []byte) (int, error) {
+	n, err := w.WriteCloser.Write(p)
+	if n > 0 && w.collector != nil {
+		w.collector.RecordQuicBytesSent(uint64(n))
+	}
+
+	return n, err
+}

network/quicstream/server.go

@@ -165,6 +165,8 @@ func (srv *Server) handleStream(ctx context.Context, remoteAddr net.Addr, stream
 	sctx, cancel := srv.streamTimeoutContext(ctx)
 	defer cancel()
 
+	r, w := wrapMetricsIO(ctx, stream, stream)
+
 	if collector := GetMetricsCollector(ctx); collector != nil {
 		collector.RecordQuicStreamOpened()
 		defer collector.RecordQuicStreamClosed()
@@ -173,7 +175,7 @@ func (srv *Server) handleStream(ctx context.Context, remoteAddr net.Addr, stream
 	var errcode quic.StreamErrorCode
 
 	if err := util.AwareContext(sctx, func(context.Context) error {
-		_, err := srv.handler(sctx, remoteAddr, stream, stream)
+		_, err := srv.handler(sctx, remoteAddr, r, w)
 
 		return err
 	}); err != nil {