From 8a8a1e53da40d4be1d70ef55f6d4bff9df0b7bb9 Mon Sep 17 00:00:00 2001
From: capossele <angelocapossele@gmail.com>
Date: Wed, 17 Jun 2020 23:24:44 +0100
Subject: [PATCH] =?UTF-8?q?=F0=9F=93=88=20Add=20clients=20metrics=20collec?=
=?UTF-8?q?tion=20via=20the=20analysis=20server?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
plugins/analysis/client/fpc_heartbeato.go | 96 ++++++++++++
plugins/analysis/client/heartbeat.go | 79 ++++++++++
plugins/analysis/client/metric_heartbeat.go | 56 +++++++
plugins/analysis/client/plugin.go | 158 +-------------------
plugins/analysis/packet/packet.go | 1 +
plugins/analysis/server/events.go | 7 +
plugins/analysis/server/plugin.go | 16 +-
plugins/metrics/parameters.go | 18 ++-
plugins/metrics/plugin.go | 58 ++++---
plugins/metrics/server.go | 57 +++++++
plugins/prometheus/analysis_server.go | 1 +
tools/docker-network/docker-compose.yml | 4 +-
12 files changed, 369 insertions(+), 182 deletions(-)
create mode 100644 plugins/analysis/client/fpc_heartbeato.go
create mode 100644 plugins/analysis/client/heartbeat.go
create mode 100644 plugins/analysis/client/metric_heartbeat.go
create mode 100644 plugins/metrics/server.go
create mode 100644 plugins/prometheus/analysis_server.go
diff --git a/plugins/analysis/client/fpc_heartbeato.go b/plugins/analysis/client/fpc_heartbeato.go
new file mode 100644
index 00000000..1cedde60
--- /dev/null
+++ b/plugins/analysis/client/fpc_heartbeato.go
@@ -0,0 +1,96 @@
+package client
+
+import (
+ "sync"
+
+ "github.com/iotaledger/goshimmer/packages/metrics"
+ "github.com/iotaledger/goshimmer/packages/vote"
+ "github.com/iotaledger/goshimmer/plugins/analysis/packet"
+ "github.com/iotaledger/goshimmer/plugins/autopeering/local"
+)
+
+var (
+ finalized map[string]vote.Opinion
+ finalizedMutex sync.RWMutex
+)
+
+func onFinalized(id string, opinion vote.Opinion) {
+ finalizedMutex.Lock()
+ finalized[id] = opinion
+ finalizedMutex.Unlock()
+}
+
+func onRoundExecuted(roundStats *vote.RoundStats) {
+ // get own ID
+ var nodeID []byte
+ if local.GetInstance() != nil {
+ // doesn't copy the ID, take care not to modify underlying bytearray!
+ nodeID = local.GetInstance().ID().Bytes()
+ }
+
+ chunks := splitFPCVoteContext(roundStats.ActiveVoteContexts)
+
+ connLock.Lock()
+ defer connLock.Unlock()
+
+ for _, chunk := range chunks {
+ // abort if empty round
+ if len(chunk) == 0 {
+ return
+ }
+
+ rs := vote.RoundStats{
+ Duration: roundStats.Duration,
+ RandUsed: roundStats.RandUsed,
+ ActiveVoteContexts: chunk,
+ }
+
+ hb := &packet.FPCHeartbeat{
+ OwnID: nodeID,
+ RoundStats: rs,
+ }
+
+ finalizedMutex.Lock()
+ hb.Finalized = finalized
+ finalized = make(map[string]vote.Opinion)
+ finalizedMutex.Unlock()
+
+ data, err := packet.NewFPCHeartbeatMessage(hb)
+ if err != nil {
+ log.Info(err, " - FPC heartbeat message skipped")
+ return
+ }
+
+ log.Info("Client: onRoundExecuted data size: ", len(data))
+
+ if _, err = managedConn.Write(data); err != nil {
+ log.Debugw("Error while writing to connection", "Description", err)
+ return
+ }
+ // trigger AnalysisOutboundBytes event
+ metrics.Events().AnalysisOutboundBytes.Trigger(uint64(len(data)))
+ }
+}
+
+func splitFPCVoteContext(ctx map[string]*vote.Context) (chunk []map[string]*vote.Context) {
+ chunk = make([]map[string]*vote.Context, 1)
+ i, counter := 0, 0
+ chunk[i] = make(map[string]*vote.Context)
+
+ if len(ctx) < maxVoteContext {
+ chunk[i] = ctx
+ return
+ }
+
+ for conflictID, voteCtx := range ctx {
+ counter++
+ if counter >= maxVoteContext {
+ counter = 0
+ i++
+ chunk = append(chunk, make(map[string]*vote.Context))
+ }
+ chunk[i][conflictID] = voteCtx
+ }
+
+ return
+}
diff --git a/plugins/analysis/client/heartbeat.go b/plugins/analysis/client/heartbeat.go
new file mode 100644
index 00000000..365562bc
--- /dev/null
+++ b/plugins/analysis/client/heartbeat.go
@@ -0,0 +1,79 @@
+package client
+
+import (
+ "strings"
+
+ "github.com/iotaledger/goshimmer/packages/metrics"
+ "github.com/iotaledger/goshimmer/plugins/analysis/packet"
+ "github.com/iotaledger/goshimmer/plugins/autopeering"
+ "github.com/iotaledger/goshimmer/plugins/autopeering/local"
+ "github.com/iotaledger/hive.go/network"
+ "github.com/mr-tron/base58"
+)
+
+// EventDispatchers holds the Heartbeat function.
+type EventDispatchers struct {
+ // Heartbeat defines the Heartbeat function.
+ Heartbeat func(heartbeat *packet.Heartbeat)
+}
+
+func sendHeartbeat(conn *network.ManagedConnection, hb *packet.Heartbeat) {
+ var out strings.Builder
+ for _, value := range hb.OutboundIDs {
+ out.WriteString(base58.Encode(value))
+ }
+ var in strings.Builder
+ for _, value := range hb.InboundIDs {
+ in.WriteString(base58.Encode(value))
+ }
+ log.Debugw(
+ "Heartbeat",
+ "nodeID", base58.Encode(hb.OwnID),
+ "outboundIDs", out.String(),
+ "inboundIDs", in.String(),
+ )
+
+ data, err := packet.NewHeartbeatMessage(hb)
+ if err != nil {
+ log.Info(err, " - heartbeat message skipped")
+ return
+ }
+
+ connLock.Lock()
+ defer connLock.Unlock()
+ if _, err = conn.Write(data); err != nil {
+ log.Debugw("Error while writing to connection", "Description", err)
+ }
+ // trigger AnalysisOutboundBytes event
+ metrics.Events().AnalysisOutboundBytes.Trigger(uint64(len(data)))
+}
+
+func createHeartbeat() *packet.Heartbeat {
+ // get own ID
+ var nodeID []byte
+ if local.GetInstance() != nil {
+ // doesn't copy the ID, take care not to modify underlying bytearray!
+ nodeID = local.GetInstance().ID().Bytes()
+ }
+
+ var outboundIDs [][]byte
+ var inboundIDs [][]byte
+
+ // get outboundIDs (chosen neighbors)
+ outgoingNeighbors := autopeering.Selection().GetOutgoingNeighbors()
+ outboundIDs = make([][]byte, len(outgoingNeighbors))
+ for i, neighbor := range outgoingNeighbors {
+ // doesn't copy the ID, take care not to modify underlying bytearray!
+ outboundIDs[i] = neighbor.ID().Bytes()
+ }
+
+ // get inboundIDs (accepted neighbors)
+ incomingNeighbors := autopeering.Selection().GetIncomingNeighbors()
+ inboundIDs = make([][]byte, len(incomingNeighbors))
+ for i, neighbor := range incomingNeighbors {
+ // doesn't copy the ID, take care not to modify underlying bytearray!
+ inboundIDs[i] = neighbor.ID().Bytes()
+ }
+
+ return &packet.Heartbeat{OwnID: nodeID, OutboundIDs: outboundIDs, InboundIDs: inboundIDs}
+}
diff --git a/plugins/analysis/client/metric_heartbeat.go b/plugins/analysis/client/metric_heartbeat.go
new file mode 100644
index 00000000..46d06ef1
--- /dev/null
+++ b/plugins/analysis/client/metric_heartbeat.go
@@ -0,0 +1,56 @@
+package client
+
+import (
+ "runtime"
+ "time"
+
+ "github.com/iotaledger/goshimmer/packages/metrics"
+ "github.com/iotaledger/goshimmer/plugins/analysis/packet"
+ "github.com/iotaledger/goshimmer/plugins/autopeering/local"
+ "github.com/iotaledger/hive.go/network"
+ "github.com/shirou/gopsutil/cpu"
+)
+
+func sendMetricHeartbeat(conn *network.ManagedConnection, hb *packet.MetricHeartbeat) {
+ data, err := packet.NewMetricHeartbeatMessage(hb)
+ if err != nil {
+ log.Info(err, " - metric heartbeat message skipped")
+ return
+ }
+
+ connLock.Lock()
+ defer connLock.Unlock()
+ if _, err = conn.Write(data); err != nil {
+ log.Debugw("Error while writing to connection", "Description", err)
+ }
+ // trigger AnalysisOutboundBytes event
+ metrics.Events().AnalysisOutboundBytes.Trigger(uint64(len(data)))
+}
+
+func createMetricHeartbeat() *packet.MetricHeartbeat {
+ // get own ID
+ var nodeID []byte
+ if local.GetInstance() != nil {
+ // doesn't copy the ID, take care not to modify underlying bytearray!
+ nodeID = local.GetInstance().ID().Bytes()
+ }
+
+ return &packet.MetricHeartbeat{
+ OwnID: nodeID,
+ OS: runtime.GOOS,
+ Arch: runtime.GOARCH,
+ NumCPU: runtime.NumCPU(),
+ CPUUsage: func() (p float64) {
+ percent, err := cpu.Percent(time.Second, false)
+ if err == nil {
+ p = percent[0]
+ }
+ return
+ }(),
+ MemoryUsage: func() uint64 {
+ var m runtime.MemStats
+ runtime.ReadMemStats(&m)
+ return m.Alloc
+ }(),
+ }
+}
diff --git a/plugins/analysis/client/plugin.go b/plugins/analysis/client/plugin.go
index b710888d..e79fc801 100644
--- a/plugins/analysis/client/plugin.go
+++ b/plugins/analysis/client/plugin.go
@@ -2,24 +2,18 @@ package client
import (
"net"
- "strings"
"sync"
"time"
"github.com/iotaledger/goshimmer/dapps/valuetransfers"
- "github.com/iotaledger/goshimmer/packages/metrics"
"github.com/iotaledger/goshimmer/packages/shutdown"
"github.com/iotaledger/goshimmer/packages/vote"
- "github.com/iotaledger/goshimmer/plugins/analysis/packet"
- "github.com/iotaledger/goshimmer/plugins/autopeering"
- "github.com/iotaledger/goshimmer/plugins/autopeering/local"
"github.com/iotaledger/goshimmer/plugins/config"
"github.com/iotaledger/hive.go/daemon"
"github.com/iotaledger/hive.go/events"
"github.com/iotaledger/hive.go/logger"
"github.com/iotaledger/hive.go/network"
"github.com/iotaledger/hive.go/node"
- "github.com/mr-tron/base58"
flag "github.com/spf13/pflag"
)
@@ -44,9 +38,6 @@ var (
log *logger.Logger
managedConn *network.ManagedConnection
connLock sync.Mutex
-
- finalized map[string]vote.Opinion
- finalizedMutex sync.RWMutex
)
func run(_ *node.Plugin) {
@@ -80,157 +71,10 @@ func run(_ *node.Plugin) {
case <-ticker.C:
sendHeartbeat(managedConn, createHeartbeat())
+ sendMetricHeartbeat(managedConn, createMetricHeartbeat())
}
}
}, shutdown.PriorityAnalysis); err != nil {
log.Panicf("Failed to start as daemon: %s", err)
}
}
-
-func onFinalized(id string, opinion vote.Opinion) {
- finalizedMutex.Lock()
- finalized[id] = opinion
- finalizedMutex.Unlock()
-}
-
-// EventDispatchers holds the Heartbeat function.
-type EventDispatchers struct {
- // Heartbeat defines the Heartbeat function.
- Heartbeat func(heartbeat *packet.Heartbeat)
-}
-
-func sendHeartbeat(conn *network.ManagedConnection, hb *packet.Heartbeat) {
- var out strings.Builder
- for _, value := range hb.OutboundIDs {
- out.WriteString(base58.Encode(value))
- }
- var in strings.Builder
- for _, value := range hb.InboundIDs {
- in.WriteString(base58.Encode(value))
- }
- log.Debugw(
- "Heartbeat",
- "nodeID", base58.Encode(hb.OwnID),
- "outboundIDs", out.String(),
- "inboundIDs", in.String(),
- )
-
- data, err := packet.NewHeartbeatMessage(hb)
- if err != nil {
- log.Info(err, " - heartbeat message skipped")
- return
- }
-
- connLock.Lock()
- defer connLock.Unlock()
- if _, err = conn.Write(data); err != nil {
- log.Debugw("Error while writing to connection", "Description", err)
- }
- // trigger AnalysisOutboundBytes event
- metrics.Events().AnalysisOutboundBytes.Trigger(uint64(len(data)))
-}
-
-func createHeartbeat() *packet.Heartbeat {
- // get own ID
- var nodeID []byte
- if local.GetInstance() != nil {
- // doesn't copy the ID, take care not to modify underlying bytearray!
- nodeID = local.GetInstance().ID().Bytes()
- }
-
- var outboundIDs [][]byte
- var inboundIDs [][]byte
-
- // get outboundIDs (chosen neighbors)
- outgoingNeighbors := autopeering.Selection().GetOutgoingNeighbors()
- outboundIDs = make([][]byte, len(outgoingNeighbors))
- for i, neighbor := range outgoingNeighbors {
- // doesn't copy the ID, take care not to modify underlying bytearray!
- outboundIDs[i] = neighbor.ID().Bytes()
- }
-
- // get inboundIDs (accepted neighbors)
- incomingNeighbors := autopeering.Selection().GetIncomingNeighbors()
- inboundIDs = make([][]byte, len(incomingNeighbors))
- for i, neighbor := range incomingNeighbors {
- // doesn't copy the ID, take care not to modify underlying bytearray!
- inboundIDs[i] = neighbor.ID().Bytes()
- }
-
- return &packet.Heartbeat{OwnID: nodeID, OutboundIDs: outboundIDs, InboundIDs: inboundIDs}
-}
-
-func onRoundExecuted(roundStats *vote.RoundStats) {
- // get own ID
- var nodeID []byte
- if local.GetInstance() != nil {
- // doesn't copy the ID, take care not to modify underlying bytearray!
- nodeID = local.GetInstance().ID().Bytes()
- }
-
- chunks := splitFPCVoteContext(roundStats.ActiveVoteContexts)
-
- connLock.Lock()
- defer connLock.Unlock()
-
- for _, chunk := range chunks {
- // abort if empty round
- if len(chunk) == 0 {
- return
- }
-
- rs := vote.RoundStats{
- Duration: roundStats.Duration,
- RandUsed: roundStats.RandUsed,
- ActiveVoteContexts: chunk,
- }
-
- hb := &packet.FPCHeartbeat{
- OwnID: nodeID,
- RoundStats: rs,
- }
-
- finalizedMutex.Lock()
- hb.Finalized = finalized
- finalized = make(map[string]vote.Opinion)
- finalizedMutex.Unlock()
-
- data, err := packet.NewFPCHeartbeatMessage(hb)
- if err != nil {
- log.Info(err, " - FPC heartbeat message skipped")
- return
- }
-
- log.Info("Client: onRoundExecuted data size: ", len(data))
-
- if _, err = managedConn.Write(data); err != nil {
- log.Debugw("Error while writing to connection", "Description", err)
- return
- }
- // trigger AnalysisOutboundBytes event
- metrics.Events().AnalysisOutboundBytes.Trigger(uint64(len(data)))
- }
-}
-
-func splitFPCVoteContext(ctx map[string]*vote.Context) (chunk []map[string]*vote.Context) {
- chunk = make([]map[string]*vote.Context, 1)
- i, counter := 0, 0
- chunk[i] = make(map[string]*vote.Context)
-
- if len(ctx) < maxVoteContext {
- chunk[i] = ctx
- return
- }
-
- for conflictID, voteCtx := range ctx {
- counter++
- if counter >= maxVoteContext {
- counter = 0
- i++
- chunk = append(chunk, make(map[string]*vote.Context))
- }
- chunk[i][conflictID] = voteCtx
- }
-
- return
-}
diff --git a/plugins/analysis/packet/packet.go b/plugins/analysis/packet/packet.go
index 442b1a28..d95c72d3 100644
--- a/plugins/analysis/packet/packet.go
+++ b/plugins/analysis/packet/packet.go
@@ -21,6 +21,7 @@ func init() {
tlv.HeaderMessageDefinition,
HeartbeatMessageDefinition,
FPCHeartbeatMessageDefinition,
+ MetricHeartbeatMessageDefinition,
}
AnalysisMsgRegistry = message.NewRegistry(definitions)
}
diff --git a/plugins/analysis/server/events.go b/plugins/analysis/server/events.go
index 95006ccd..75828e1f 100644
--- a/plugins/analysis/server/events.go
+++ b/plugins/analysis/server/events.go
@@ -21,6 +21,8 @@ var Events = struct {
Heartbeat *events.Event
// FPCHeartbeat triggers when an FPC heartbeat has been received.
FPCHeartbeat *events.Event
+ // MetricHeartbeat triggers when an MetricHeartbeat heartbeat has been received.
+ MetricHeartbeat *events.Event
}{
events.NewEvent(stringCaller),
events.NewEvent(stringCaller),
@@ -29,6 +31,7 @@ var Events = struct {
events.NewEvent(errorCaller),
events.NewEvent(heartbeatPacketCaller),
events.NewEvent(fpcHeartbeatPacketCaller),
+ events.NewEvent(metricHeartbeatPacketCaller),
}
func stringCaller(handler interface{}, params ...interface{}) {
@@ -50,3 +53,7 @@ func heartbeatPacketCaller(handler interface{}, params ...interface{}) {
func fpcHeartbeatPacketCaller(handler interface{}, params ...interface{}) {
handler.(func(hb *packet.FPCHeartbeat))(params[0].(*packet.FPCHeartbeat))
}
+
+func metricHeartbeatPacketCaller(handler interface{}, params ...interface{}) {
+ handler.(func(hb *packet.MetricHeartbeat))(params[0].(*packet.MetricHeartbeat))
+}
diff --git a/plugins/analysis/server/plugin.go b/plugins/analysis/server/plugin.go
index fcd6c232..30e317a2 100644
--- a/plugins/analysis/server/plugin.go
+++ b/plugins/analysis/server/plugin.go
@@ -112,6 +112,9 @@ func wireUp(p *protocol.Protocol) {
p.Events.Received[packet.MessageTypeFPCHeartbeat].Attach(events.NewClosure(func(data []byte) {
processFPCHeartbeatPacket(data, p)
}))
+ p.Events.Received[packet.MessageTypeMetricHeartbeat].Attach(events.NewClosure(func(data []byte) {
+ processMetricHeartbeatPacket(data, p)
+ }))
}
// processHeartbeatPacket parses the serialized data into a Heartbeat packet and triggers its event
@@ -125,7 +128,7 @@ func processHeartbeatPacket(data []byte, p *protocol.Protocol) {
Events.Heartbeat.Trigger(heartbeatPacket)
}
-// processHeartbeatPacket parses the serialized data into a Heartbeat packet and triggers its event
+// processHeartbeatPacket parses the serialized data into a FPC Heartbeat packet and triggers its event
func processFPCHeartbeatPacket(data []byte, p *protocol.Protocol) {
hb, err := packet.ParseFPCHeartbeat(data)
if err != nil {
@@ -135,3 +138,14 @@ func processFPCHeartbeatPacket(data []byte, p *protocol.Protocol) {
}
Events.FPCHeartbeat.Trigger(hb)
}
+
+// processMetricHeartbeatPacket parses the serialized data into a MEtric Heartbeat packet and triggers its event
+func processMetricHeartbeatPacket(data []byte, p *protocol.Protocol) {
+ hb, err := packet.ParseMetricHeartbeat(data)
+ if err != nil {
+ Events.Error.Trigger(err)
+ p.CloseConnection()
+ return
+ }
+ Events.MetricHeartbeat.Trigger(hb)
+}
diff --git a/plugins/metrics/parameters.go b/plugins/metrics/parameters.go
index 31df451f..b504f77e 100644
--- a/plugins/metrics/parameters.go
+++ b/plugins/metrics/parameters.go
@@ -1,6 +1,10 @@
package metrics
-import "time"
+import (
+ "time"
+
+ flag "github.com/spf13/pflag"
+)
const (
// should always be 1 second
@@ -13,3 +17,15 @@ const (
MemUsageMeasurementInterval = 1 * time.Second
SyncedMeasurementInterval = 1 * time.Second
)
+
+const (
+ // CfgMetricsLocal defines the config flag to enable/disable local metrics.
+ CfgMetricsLocal = "metrics.local"
+ // CfgMetricsGlobal defines the config flag to enable/disable global metrics.
+ CfgMetricsGlobal = "metrics.global"
+)
+
+func init() {
+ flag.Bool(CfgMetricsLocal, true, "include local metrics")
+ flag.Bool(CfgMetricsGlobal, false, "include global metrics")
+}
diff --git a/plugins/metrics/plugin.go b/plugins/metrics/plugin.go
index ca67ba43..608dd9a6 100644
--- a/plugins/metrics/plugin.go
+++ b/plugins/metrics/plugin.go
@@ -12,7 +12,9 @@ import (
"github.com/iotaledger/goshimmer/packages/metrics"
"github.com/iotaledger/goshimmer/packages/shutdown"
"github.com/iotaledger/goshimmer/packages/vote"
+ "github.com/iotaledger/goshimmer/plugins/analysis/server"
"github.com/iotaledger/goshimmer/plugins/autopeering"
+ "github.com/iotaledger/goshimmer/plugins/config"
"github.com/iotaledger/goshimmer/plugins/gossip"
"github.com/iotaledger/goshimmer/plugins/messagelayer"
"github.com/iotaledger/hive.go/daemon"
@@ -36,6 +38,40 @@ func configure(_ *node.Plugin) {
func run(_ *node.Plugin) {
+ if config.Node.GetBool(CfgMetricsLocal) {
+ registerLocalMetrics()
+ }
+
+ // Events from analysis server
+ if config.Node.GetBool(CfgMetricsGlobal) {
+ server.Events.MetricHeartbeat.Attach(onMetricHeartbeatReceived)
+ }
+
+ // create a background worker that update the metrics every second
+ if err := daemon.BackgroundWorker("Metrics Updater", func(shutdownSignal <-chan struct{}) {
+ if config.Node.GetBool(CfgMetricsLocal) {
+ timeutil.Ticker(func() {
+ measureReceivedMPS()
+ measureCPUUsage()
+ measureMemUsage()
+ measureSynced()
+
+ // gossip network traffic
+ g := gossipCurrentTraffic()
+ gossipCurrentRx.Store(uint64(g.BytesRead))
+ gossipCurrentTx.Store(uint64(g.BytesWritten))
+ }, 1*time.Second, shutdownSignal)
+ timeutil.Ticker(measureMPSPerPayload, MPSMeasurementInterval, shutdownSignal)
+ timeutil.Ticker(measureMessageTips, MessageTipsMeasurementInterval, shutdownSignal)
+ timeutil.Ticker(measureReceivedTPS, TPSMeasurementInterval, shutdownSignal)
+ timeutil.Ticker(measureValueTips, ValueTipsMeasurementInterval, shutdownSignal)
+ }
+ }, shutdown.PriorityMetrics); err != nil {
+ log.Panicf("Failed to start as daemon: %s", err)
+ }
+}
+
+func registerLocalMetrics() {
//// Events declared in other packages which we want to listen to here ////
// increase received MPS counter whenever we attached a message
@@ -110,26 +146,4 @@ func run(_ *node.Plugin) {
metrics.Events().QueryReplyError.Attach(events.NewClosure(func(ev *metrics.QueryReplyErrorEvent) {
processQueryReplyError(ev)
}))
-
- // create a background worker that "measures" the MPS value every second
- if err := daemon.BackgroundWorker("Metrics Updater", func(shutdownSignal <-chan struct{}) {
- timeutil.Ticker(func() {
- measureReceivedMPS()
- measureCPUUsage()
- measureMemUsage()
- measureSynced()
-
- // gossip network traffic
- g := gossipCurrentTraffic()
- gossipCurrentRx.Store(uint64(g.BytesRead))
- gossipCurrentTx.Store(uint64(g.BytesWritten))
-
- }, 1*time.Second, shutdownSignal)
- timeutil.Ticker(measureMPSPerPayload, MPSMeasurementInterval, shutdownSignal)
- timeutil.Ticker(measureMessageTips, MessageTipsMeasurementInterval, shutdownSignal)
- timeutil.Ticker(measureReceivedTPS, TPSMeasurementInterval, shutdownSignal)
- timeutil.Ticker(measureValueTips, ValueTipsMeasurementInterval, shutdownSignal)
- }, shutdown.PriorityMetrics); err != nil {
- log.Panicf("Failed to start as daemon: %s", err)
- }
}
diff --git a/plugins/metrics/server.go b/plugins/metrics/server.go
new file mode 100644
index 00000000..3c3bc378
--- /dev/null
+++ b/plugins/metrics/server.go
@@ -0,0 +1,57 @@
+package metrics
+
+import (
+ "bytes"
+ "encoding/gob"
+ "sync"
+
+ "github.com/iotaledger/goshimmer/plugins/analysis/packet"
+ "github.com/iotaledger/hive.go/events"
+ "github.com/mr-tron/base58/base58"
+)
+
+type ClientInfo struct {
+ OS string
+ Arch string
+ NumCPU int
+ CPUUsage float64
+ MemoryUsage uint64
+}
+
+var (
+ clientsMetrics = make(map[string]ClientInfo)
+ clientsMetricsMutex sync.RWMutex
+)
+
+var onMetricHeartbeatReceived = events.NewClosure(func(hb *packet.MetricHeartbeat) {
+ clientsMetricsMutex.Lock()
+ defer clientsMetricsMutex.Unlock()
+ clientsMetrics[base58.Encode(hb.OwnID)] = ClientInfo{
+ OS: hb.OS,
+ Arch: hb.Arch,
+ NumCPU: hb.NumCPU,
+ CPUUsage: hb.CPUUsage,
+ MemoryUsage: hb.MemoryUsage,
+ }
+})
+
+func ClientsMetrics() map[string]ClientInfo {
+ clientsMetricsMutex.RLock()
+ defer clientsMetricsMutex.RUnlock()
+
+ var buf bytes.Buffer
+ enc := gob.NewEncoder(&buf)
+ err := enc.Encode(clientsMetrics)
+ if err != nil {
+ return nil
+ }
+
+ dec := gob.NewDecoder(&buf)
+ var copy map[string]ClientInfo
+ err = dec.Decode(©)
+ if err != nil {
+ return nil
+ }
+
+ return copy
+}
diff --git a/plugins/prometheus/analysis_server.go b/plugins/prometheus/analysis_server.go
new file mode 100644
index 00000000..7b1b4c03
--- /dev/null
+++ b/plugins/prometheus/analysis_server.go
@@ -0,0 +1 @@
+package prometheus
diff --git a/tools/docker-network/docker-compose.yml b/tools/docker-network/docker-compose.yml
index 467efcc7..6763e418 100644
--- a/tools/docker-network/docker-compose.yml
+++ b/tools/docker-network/docker-compose.yml
@@ -24,7 +24,9 @@ services:
--analysis.dashboard.bindAddress=0.0.0.0:9000
--node.enablePlugins=analysis-server,analysis-dashboard
--analysis.dashboard.dev=false
- --node.disablePlugins=portcheck,dashboard,analysis-client,gossip,drng,issuer,sync,metrics,messagelayer,valuetransfers,webapi,webapibroadcastdataendpoint,webapifindtransactionhashesendpoint,webapigetneighborsendpoint,webapigettransactionobjectsbyhashendpoint,webapigettransactiontrytesbyhashendpoint
+ --metrics.local=false
+ --metrics.global=true
+ --node.disablePlugins=portcheck,dashboard,analysis-client,gossip,drng,issuer,sync,messagelayer,valuetransfers,webapi,webapibroadcastdataendpoint,webapifindtransactionhashesendpoint,webapigetneighborsendpoint,webapigettransactionobjectsbyhashendpoint,webapigettransactiontrytesbyhashendpoint
volumes:
- ./config.docker.json:/tmp/config.json:ro
- goshimmer-cache:/go
--
GitLab