plugin.go

package metrics

import (
	"sync"
	"time"

	"github.com/iotaledger/hive.go/daemon"
	"github.com/iotaledger/hive.go/events"
	"github.com/iotaledger/hive.go/logger"
	"github.com/iotaledger/hive.go/node"
	"github.com/iotaledger/hive.go/timeutil"

	gossippkg "github.com/iotaledger/goshimmer/packages/gossip"
	"github.com/iotaledger/goshimmer/packages/mana"
	"github.com/iotaledger/goshimmer/packages/metrics"
	"github.com/iotaledger/goshimmer/packages/shutdown"
	"github.com/iotaledger/goshimmer/packages/tangle"
	"github.com/iotaledger/goshimmer/packages/vote"
	"github.com/iotaledger/goshimmer/plugins/analysis/server"
	"github.com/iotaledger/goshimmer/plugins/autopeering"
	"github.com/iotaledger/goshimmer/plugins/gossip"
	"github.com/iotaledger/goshimmer/plugins/messagelayer"
)

// PluginName is the name of the metrics plugin.
const PluginName = "Metrics"

var (
	// plugin is the plugin instance of the metrics plugin.
	plugin *node.Plugin
	once   sync.Once
	log    *logger.Logger
)

// Plugin gets the plugin instance.
func Plugin() *node.Plugin {
	once.Do(func() {
		plugin = node.NewPlugin(PluginName, node.Enabled, configure, run)
	})
	return plugin
}

func configure(_ *node.Plugin) {
	log = logger.NewLogger(PluginName)
}

func run(_ *node.Plugin) {
	log.Infof("Starting %s ...", PluginName)
	if Parameters.Local {
		// initial measurement, since we have to know how many messages are there in the db
		measureInitialDBStats()
		registerLocalMetrics()
	}
	// Events from analysis server
	if Parameters.Global {
		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 Parameters.Local {
			// Do not block until the Ticker is shutdown because we might want to start multiple Tickers and we can
			// safely ignore the last execution when shutting down.
			timeutil.NewTicker(func() {
				measureCPUUsage()
				measureMemUsage()
				measureSynced()
				measureMessageTips()
				measureReceivedMPS()
				measureRequestQueueSize()
				measureGossipTraffic()
				measurePerComponentCounter()
			}, 1*time.Second, shutdownSignal)
		}

		if Parameters.Global {
			// Do not block until the Ticker is shutdown because we might want to start multiple Tickers and we can
			// safely ignore the last execution when shutting down.
			timeutil.NewTicker(calculateNetworkDiameter, 1*time.Minute, shutdownSignal)
		}

		// Wait before terminating so we get correct log messages from the daemon regarding the shutdown order.
		<-shutdownSignal
	}, shutdown.PriorityMetrics); err != nil {
		log.Panicf("Failed to start as daemon: %s", err)
	}

	// create a background worker that updates the mana metrics
	if err := daemon.BackgroundWorker("Metrics Mana Updater", func(shutdownSignal <-chan struct{}) {
		defer log.Infof("Stopping Metrics Mana Updater ... done")
		timeutil.NewTicker(func() {
			measureMana()
		}, time.Second*time.Duration(Parameters.ManaUpdateInterval), shutdownSignal)
		// Wait before terminating so we get correct log messages from the daemon regarding the shutdown order.
		<-shutdownSignal
		log.Infof("Stopping Metrics Mana Updater ...")
	}, shutdown.PriorityMetrics); err != nil {
		log.Panicf("Failed to start as daemon: %s", err)
	}

	if Parameters.ManaResearch {
		// create a background worker that updates the research mana metrics
		if err := daemon.BackgroundWorker("Metrics Research Mana Updater", func(shutdownSignal <-chan struct{}) {
			defer log.Infof("Stopping Metrics Research Mana Updater ... done")
			timeutil.NewTicker(func() {
				measureAccessResearchMana()
				measureConsensusResearchMana()
			}, time.Second*time.Duration(Parameters.ManaUpdateInterval), shutdownSignal)
			// Wait before terminating so we get correct log messages from the daemon regarding the shutdown order.
			<-shutdownSignal
			log.Infof("Stopping Metrics Research Mana Updater ...")
		}, 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
	messagelayer.Tangle().Storage.Events.MessageStored.Attach(events.NewClosure(func(messageID tangle.MessageID) {
		messagelayer.Tangle().Storage.Message(messageID).Consume(func(message *tangle.Message) {
			increaseReceivedMPSCounter()
			increasePerPayloadCounter(message.Payload().Type())
			// MessageStored is triggered in storeMessageWorker that saves the msg to database
			messageTotalCountDB.Inc()
		})
		increasePerComponentCounter(Store)
	}))

	messagelayer.Tangle().Storage.Events.MessageRemoved.Attach(events.NewClosure(func(messageId tangle.MessageID) {
		// MessageRemoved triggered when the message gets removed from database.
		messageTotalCountDB.Dec()
	}))

	// messages can only become solid once, then they stay like that, hence no .Dec() part
	messagelayer.Tangle().Solidifier.Events.MessageSolid.Attach(events.NewClosure(func(messageID tangle.MessageID) {
		increasePerComponentCounter(Solidifier)
		solidTimeMutex.Lock()
		defer solidTimeMutex.Unlock()

		// Consume should release cachedMessageMetadata
		messagelayer.Tangle().Storage.MessageMetadata(messageID).Consume(func(msgMetaData *tangle.MessageMetadata) {
			if msgMetaData.IsSolid() {
				messageSolidCountDBInc.Inc()
				sumSolidificationTime += msgMetaData.SolidificationTime().Sub(msgMetaData.ReceivedTime())
			}
		})
	}))

	// fired when a message gets added to missing message storage
	messagelayer.Tangle().Solidifier.Events.MessageMissing.Attach(events.NewClosure(func(messageId tangle.MessageID) {
		missingMessageCountDB.Inc()
	}))

	// fired when a missing message was received and removed from missing message storage
	messagelayer.Tangle().Storage.Events.MissingMessageStored.Attach(events.NewClosure(func(tangle.MessageID) {
		missingMessageCountDB.Dec()
	}))

	messagelayer.Tangle().Scheduler.Events.MessageScheduled.Attach(events.NewClosure(func(messageID tangle.MessageID) {
		increasePerComponentCounter(Scheduler)
	}))
	messagelayer.Tangle().FIFOScheduler.Events.MessageScheduled.Attach(events.NewClosure(func(messageID tangle.MessageID) {
		increasePerComponentCounter(Scheduler)
	}))

	messagelayer.Tangle().Booker.Events.MessageBooked.Attach(events.NewClosure(func(message tangle.MessageID) {
		increasePerComponentCounter(Booker)
	}))

	// // Value payload attached
	// valuetransfers.Tangle().Events.PayloadAttached.Attach(events.NewClosure(func(cachedPayloadEvent *valuetangle.CachedPayloadEvent) {
	// 	cachedPayloadEvent.Payload.Release()
	// 	cachedPayloadEvent.PayloadMetadata.Release()
	// 	valueTransactionCounter.Inc()
	// }))

	// FPC round executed
	messagelayer.Voter().Events().RoundExecuted.Attach(events.NewClosure(func(roundStats *vote.RoundStats) {
		processRoundStats(roundStats)
	}))

	// a conflict has been finalized
	messagelayer.Voter().Events().Finalized.Attach(events.NewClosure(func(ev *vote.OpinionEvent) {
		processFinalized(ev.Ctx)
	}))

	// consensus failure in conflict resolution
	messagelayer.Voter().Events().Failed.Attach(events.NewClosure(func(ev *vote.OpinionEvent) {
		processFailed(ev.Ctx)
	}))

	//// Events coming from metrics package ////

	metrics.Events().FPCInboundBytes.Attach(events.NewClosure(func(amountBytes uint64) {
		_FPCInboundBytes.Add(amountBytes)
	}))
	metrics.Events().FPCOutboundBytes.Attach(events.NewClosure(func(amountBytes uint64) {
		_FPCOutboundBytes.Add(amountBytes)
	}))
	metrics.Events().AnalysisOutboundBytes.Attach(events.NewClosure(func(amountBytes uint64) {
		analysisOutboundBytes.Add(amountBytes)
	}))
	metrics.Events().CPUUsage.Attach(events.NewClosure(func(cpuPercent float64) {
		cpuUsage.Store(cpuPercent)
	}))
	metrics.Events().MemUsage.Attach(events.NewClosure(func(memAllocBytes uint64) {
		memUsageBytes.Store(memAllocBytes)
	}))
	metrics.Events().TangleTimeSynced.Attach(events.NewClosure(func(synced bool) {
		isTangleTimeSynced.Store(synced)
	}))

	gossip.Manager().NeighborsEvents(gossippkg.NeighborsGroupAuto).NeighborRemoved.Attach(onNeighborRemoved)
	gossip.Manager().NeighborsEvents(gossippkg.NeighborsGroupAuto).NeighborAdded.Attach(onNeighborAdded)

	autopeering.Selection().Events().IncomingPeering.Attach(onAutopeeringSelection)
	autopeering.Selection().Events().OutgoingPeering.Attach(onAutopeeringSelection)

	metrics.Events().MessageTips.Attach(events.NewClosure(func(tipsCount uint64) {
		messageTips.Store(tipsCount)
	}))

	metrics.Events().QueryReceived.Attach(events.NewClosure(func(ev *metrics.QueryReceivedEvent) {
		processQueryReceived(ev)
	}))
	metrics.Events().QueryReplyError.Attach(events.NewClosure(func(ev *metrics.QueryReplyErrorEvent) {
		processQueryReplyError(ev)
	}))

	// mana pledge events
	mana.Events().Pledged.Attach(events.NewClosure(func(ev *mana.PledgedEvent) {
		addPledge(ev)
	}))
}