Skip to content
Snippets Groups Projects
Select Git revision
  • 6d3b2a2c640102739e8370000150eb536d11ce46
  • without_tipselection default
  • develop protected
  • fix/grafana-local-dashboard
  • wasp
  • fix/dashboard-explorer-freeze
  • master
  • feat/timerqueue
  • test/sync_debug_and_650
  • feat/sync_revamp_inv
  • wip/sync
  • tool/db-recovery
  • portcheck/fix
  • fix/synchronization
  • feat/new-dashboard-analysis
  • feat/refactored-analysis-dashboard
  • feat/new-analysis-dashboard
  • test/demo-prometheus-fpc
  • prometheus_metrics
  • wip/analysis-server
  • merge/fpc-test-value-transfer
  • v0.2.2
  • v0.2.1
  • v0.2.0
  • v0.1.3
  • v0.1.2
  • v0.1.1
  • v0.1.0
28 results

plugin.go

Blame
  • user avatar
    Luca Moser authored and GitHub committed
    * normalize bind addresses prints
    
    * refactors analysis plugins
    
    * adds tests for parsing/serializing heartbeat packets
    
    * adds additional test case
    
    * normalize usage of word 'ids'
    
    * updates packr files
    
    * normalizes handler routes for static assets for the analysis web interface
    
    * normalize config key names
    
    * adjusts configs for new analysis config keys
    
    * fixes review comments
    6d3b2a2c
    History
    plugin.go 3.87 KiB
    package client
    
    import (
    	"encoding/hex"
    	"net"
    	"strings"
    	"sync"
    	"time"
    
    	"github.com/iotaledger/goshimmer/packages/shutdown"
    	"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/logger"
    	"github.com/iotaledger/hive.go/network"
    	"github.com/iotaledger/hive.go/node"
    	flag "github.com/spf13/pflag"
    )
    
    const (
    	// PluginName is the name of  the analysis client plugin.
    	PluginName = "Analysis-Client"
    	// CfgServerAddress defines the config flag of the analysis server address.
    	CfgServerAddress = "analysis.client.serverAddress"
    	// defines the report interval of the reporting in seconds.
    	reportIntervalSec = 5
    )
    
    func init() {
    	flag.String(CfgServerAddress, "ressims.iota.cafe:188", "tcp server for collecting analysis information")
    }
    
    var (
    	// Plugin is the plugin instance of the analysis client plugin.
    	Plugin   = node.NewPlugin(PluginName, node.Enabled, run)
    	log      *logger.Logger
    	connLock sync.Mutex
    )
    
    func run(_ *node.Plugin) {
    	log = logger.NewLogger(PluginName)
    	if err := daemon.BackgroundWorker(PluginName, func(shutdownSignal <-chan struct{}) {
    		ticker := time.NewTicker(reportIntervalSec * time.Second)
    		defer ticker.Stop()
    		for {
    			select {
    			case <-shutdownSignal:
    				return
    
    			case <-ticker.C:
    				conn, err := net.Dial("tcp", config.Node.GetString(CfgServerAddress))
    				if err != nil {
    					log.Debugf("Could not connect to reporting server: %s", err.Error())
    					continue
    				}
    				managedConn := network.NewManagedConnection(conn)
    				eventDispatchers := getEventDispatchers(managedConn)
    				reportHeartbeat(eventDispatchers)
    			}
    		}
    	}, shutdown.PriorityAnalysis); err != nil {
    		log.Panic(err)
    	}
    }
    
    // EventDispatchers holds the Heartbeat function.
    type EventDispatchers struct {
    	// Heartbeat defines the Heartbeat function.
    	Heartbeat func(heartbeat *packet.Heartbeat)
    }
    
    func getEventDispatchers(conn *network.ManagedConnection) *EventDispatchers {
    	return &EventDispatchers{
    		Heartbeat: func(hb *packet.Heartbeat) {
    			var out strings.Builder
    			for _, value := range hb.OutboundIDs {
    				out.WriteString(hex.EncodeToString(value))
    			}
    			var in strings.Builder
    			for _, value := range hb.InboundIDs {
    				in.WriteString(hex.EncodeToString(value))
    			}
    			log.Debugw(
    				"Heartbeat",
    				"nodeID", hex.EncodeToString(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)
    			}
    		},
    	}
    }
    
    func reportHeartbeat(dispatchers *EventDispatchers) {
    	// 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()
    	}
    
    	hb := &packet.Heartbeat{OwnID: nodeID, OutboundIDs: outboundIDs, InboundIDs: inboundIDs}
    	dispatchers.Heartbeat(hb)
    }