broadcast node's height info into p2p network

api/coreservice.go

@@ -158,6 +158,7 @@ type (
 		chainListener     apitypes.Listener
 		electionCommittee committee.Committee
 		readCache         *ReadCache
+		delegatesMonitor  map[string]*iotextypes.ConsensusMessage
 	}
 
 	// jobDesc provides a struct to get and store logs in core.LogsInRange

api/serverV2.go

@@ -21,6 +21,7 @@ import (
 	"github.com/iotexproject/iotex-core/blocksync"
 	"github.com/iotexproject/iotex-core/pkg/tracer"
 	"github.com/iotexproject/iotex-core/state/factory"
+	"github.com/iotexproject/iotex-proto/golang/iotextypes"
 )
 
 // ServerV2 provides api for user to interact with blockchain data
@@ -134,3 +135,9 @@ func (svr *ServerV2) ReceiveBlock(blk *block.Block) error {
 func (svr *ServerV2) CoreService() CoreService {
 	return svr.core
 }
+
+// HandleMonitorMsg handle monitor msg
+func (svr *ServerV2) HandleMonitorMsg(msg *iotextypes.ConsensusMessage) error {
+	// save msg to core.delegatesMonitor
+	return nil
+}

config/config.go

@@ -25,6 +25,7 @@ import (
 	"github.com/iotexproject/iotex-core/dispatcher"
 	"github.com/iotexproject/iotex-core/p2p"
 	"github.com/iotexproject/iotex-core/pkg/log"
+	"github.com/iotexproject/iotex-core/server/cronjob"
 )
 
 // IMPORTANT: to define a config, add a field or a new config type to the existing config types. In addition, provide
@@ -80,6 +81,7 @@ var (
 		DB:      db.DefaultConfig,
 		Indexer: blockindex.DefaultConfig,
 		Genesis: genesis.Default,
+		Cronjob: cronjob.DefaultConfig,
 	}
 
 	// ErrInvalidCfg indicates the invalid config value
@@ -129,6 +131,7 @@ type (
 		Log                log.GlobalConfig                `yaml:"log"`
 		SubLogs            map[string]log.GlobalConfig     `yaml:"subLogs"`
 		Genesis            genesis.Genesis                 `yaml:"genesis"`
+		Cronjob            cronjob.Config                  `yaml:"cronjob"`
 	}
 
 	// Validate is the interface of validating the config

dispatcher/dispatcher.go

@@ -55,6 +55,11 @@ type Subscriber interface {
 	HandleConsensusMsg(*iotextypes.ConsensusMessage) error
 }
 
+// MsgSubscriber is the dispatcher msg subscriber interface
+type MsgSubscriber interface {
+	HandleMonitorMsg(*iotextypes.ConsensusMessage) error
+}
+
 // Dispatcher is used by peers, handles incoming block and header notifications and relays announcements of new blocks.
 type Dispatcher interface {
 	lifecycle.StartStopper
@@ -67,18 +72,30 @@ type Dispatcher interface {
 	// HandleTell handles the incoming tell message. The transportation layer semantics is exact once. The sender is
 	// given for the sake of replying the message
 	HandleTell(context.Context, uint32, peer.AddrInfo, proto.Message)
+	// AddMsgSubscriber add msg subscriber
+	AddMsgSubscriber(MsgSubscriber)
 }
 
-var requestMtc = prometheus.NewCounterVec(
-	prometheus.CounterOpts{
-		Name: "iotex_dispatch_request",
-		Help: "Dispatcher request counter.",
-	},
-	[]string{"method", "succeed"},
+var (
+	requestMtc = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "iotex_dispatch_request",
+			Help: "Dispatcher request counter.",
+		},
+		[]string{"method", "succeed"},
+	)
+	delegateHeightGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "iotex_p2p_delegate_height_gauge",
+			Help: "delegate height",
+		},
+		[]string{"address", "version"},
+	)
 )
 
 func init() {
 	prometheus.MustRegister(requestMtc)
+	prometheus.MustRegister(delegateHeightGauge)
 }
 
 // blockMsg packages a proto block message.
@@ -118,22 +135,24 @@ func (m actionMsg) ChainID() uint32 {
 
 // IotxDispatcher is the request and event dispatcher for iotx node.
 type IotxDispatcher struct {
-	started        int32
-	shutdown       int32
-	actionChanLock sync.RWMutex
-	blockChanLock  sync.RWMutex
-	syncChanLock   sync.RWMutex
-	actionChan     chan *actionMsg
-	blockChan      chan *blockMsg
-	syncChan       chan *blockSyncMsg
-	eventAudit     map[iotexrpc.MessageType]int
-	eventAuditLock sync.RWMutex
-	wg             sync.WaitGroup
-	quit           chan struct{}
-	subscribers    map[uint32]Subscriber
-	subscribersMU  sync.RWMutex
-	peerLastSync   map[string]time.Time
-	syncInterval   time.Duration
+	started          int32
+	shutdown         int32
+	actionChanLock   sync.RWMutex
+	blockChanLock    sync.RWMutex
+	syncChanLock     sync.RWMutex
+	actionChan       chan *actionMsg
+	blockChan        chan *blockMsg
+	syncChan         chan *blockSyncMsg
+	eventAudit       map[iotexrpc.MessageType]int
+	eventAuditLock   sync.RWMutex
+	wg               sync.WaitGroup
+	quit             chan struct{}
+	subscribers      map[uint32]Subscriber
+	subscribersMU    sync.RWMutex
+	peerLastSync     map[string]time.Time
+	syncInterval     time.Duration
+	msgSubscribers   []MsgSubscriber
+	msgSubscribersMU sync.RWMutex
 }
 
 // NewDispatcher creates a new Dispatcher
@@ -161,6 +180,15 @@ func (d *IotxDispatcher) AddSubscriber(
 	d.subscribersMU.Unlock()
 }
 
+// AddMsgSubscriber adds a subscriber to dispatcher
+func (d *IotxDispatcher) AddMsgSubscriber(
+	subscriber MsgSubscriber,
+) {
+	d.msgSubscribersMU.Lock()
+	d.msgSubscribers = append(d.msgSubscribers, subscriber)
+	d.msgSubscribersMU.Unlock()
+}
+
 // Start starts the dispatcher.
 func (d *IotxDispatcher) Start(ctx context.Context) error {
 	if atomic.AddInt32(&d.started, 1) != 1 {
@@ -345,6 +373,10 @@ func (d *IotxDispatcher) dispatchAction(ctx context.Context, chainID uint32, msg
 	subscriber.ReportFullness(ctx, iotexrpc.MessageType_ACTION, float32(l)/float32(c))
 }
 
+func (d *IotxDispatcher) dispatchMonitorMessage(ctx context.Context, peer string, msg proto.Message) {
+	// save metrics
+}
+
 // dispatchBlock adds the passed block message to the news handling queue.
 func (d *IotxDispatcher) dispatchBlock(ctx context.Context, chainID uint32, peer string, msg proto.Message) {
 	if atomic.LoadInt32(&d.shutdown) != 0 {
@@ -408,8 +440,22 @@ func (d *IotxDispatcher) dispatchBlockSyncReq(ctx context.Context, chainID uint3
 	subscriber.ReportFullness(ctx, iotexrpc.MessageType_BLOCK_REQUEST, float32(l)/float32(c))
 }
 
+func (d *IotxDispatcher) handleBroadcastMsg(ctx context.Context, chainID uint32, peer string, message proto.Message) {
+	for _, sub := range d.msgSubscribers {
+		switch msg := message.(type) {
+		case *iotextypes.ConsensusMessage:
+			if err := sub.HandleMonitorMsg(msg); err != nil {
+				log.L().Debug("Failed to handle consensus message", zap.Error(err))
+			}
+		default:
+		}
+	}
+}
+
 // HandleBroadcast handles incoming broadcast message
 func (d *IotxDispatcher) HandleBroadcast(ctx context.Context, chainID uint32, peer string, message proto.Message) {
+	d.handleBroadcastMsg(ctx, chainID, peer, message)
+
 	subscriber := d.subscriber(chainID)
 	if subscriber == nil {
 		log.L().Warn("chainID has not been registered in dispatcher.", zap.Uint32("chainID", chainID))
@@ -425,6 +471,8 @@ func (d *IotxDispatcher) HandleBroadcast(ctx context.Context, chainID uint32, pe
 		d.dispatchAction(ctx, chainID, message)
 	case *iotextypes.Block:
 		d.dispatchBlock(ctx, chainID, peer, message)
+	case *iotextypes.ConsensusVote:
+		d.dispatchMonitorMessage(ctx, peer, msg)
 	default:
 		msgType, _ := goproto.GetTypeFromRPCMsg(message)
 		log.L().Warn("Unexpected msgType handled by HandleBroadcast.", zap.Any("msgType", msgType))

server/cronjob/config.go

@@ -0,0 +1,18 @@
+// Copyright (c) 2022 IoTeX Foundation
+// This source code is provided 'as is' and no warranties are given as to title or non-infringement, merchantability
+// or fitness for purpose and, to the extent permitted by law, all liability for your use of the code is disclaimed.
+// This source code is governed by Apache License 2.0 that can be found in the LICENSE file.
+
+package cronjob
+
+import "time"
+
+// Config is the cronjob config
+type Config struct {
+	MonitorInterval time.Duration `yaml:"monitorInterval"`
+}
+
+// DefaultConfig is the default config
+var DefaultConfig = Config{
+	MonitorInterval: 5 * time.Minute,
+}

server/cronjob/monitor.go

@@ -0,0 +1,58 @@
+// Copyright (c) 2022 IoTeX Foundation
+// This source code is provided 'as is' and no warranties are given as to title or non-infringement, merchantability
+// or fitness for purpose and, to the extent permitted by law, all liability for your use of the code is disclaimed.
+// This source code is governed by Apache License 2.0 that can be found in the LICENSE file.
+
+package cronjob
+
+import (
+	"context"
+	"time"
+
+	"github.com/iotexproject/iotex-core/blockchain"
+	"github.com/iotexproject/iotex-core/p2p"
+	"github.com/iotexproject/iotex-proto/golang/iotextypes"
+)
+
+// CronJob interface
+type CronJob interface {
+	Run()
+	Interval() time.Duration
+}
+
+type cronInterval time.Duration
+
+func (i cronInterval) Interval() time.Duration {
+	return time.Duration(i)
+}
+
+type monitorJob struct {
+	cronInterval
+	agent p2p.Agent
+	bc    blockchain.Blockchain
+}
+
+// NewMonitorCronjob new monitor cronjob
+func NewMonitorCronjob(agent p2p.Agent, bc blockchain.Blockchain, interval time.Duration) CronJob {
+	return &monitorJob{
+		cronInterval: cronInterval(interval),
+		agent:        agent,
+		bc:           bc,
+	}
+}
+
+func (j *monitorJob) Run() {
+	// ver := version.PackageVersion
+	j.agent.BroadcastOutbound(context.Background(), &iotextypes.ConsensusMessage{
+		Height: j.bc.TipHeight(),
+	})
+}
+
+// NewCronJobs create cronjobs
+func NewCronJobs(cfg Config, agent p2p.Agent, bc blockchain.Blockchain) []CronJob {
+	res := []CronJob{}
+	if cfg.MonitorInterval > 0 {
+		res = append(res, NewMonitorCronjob(agent, bc, cfg.MonitorInterval))
+	}
+	return res
+}

server/itx/server.go

@@ -22,10 +22,12 @@ import (
 	"github.com/iotexproject/iotex-core/dispatcher"
 	"github.com/iotexproject/iotex-core/p2p"
 	"github.com/iotexproject/iotex-core/pkg/ha"
+	"github.com/iotexproject/iotex-core/pkg/lifecycle"
 	"github.com/iotexproject/iotex-core/pkg/log"
 	"github.com/iotexproject/iotex-core/pkg/probe"
 	"github.com/iotexproject/iotex-core/pkg/routine"
 	"github.com/iotexproject/iotex-core/pkg/util/httputil"
+	"github.com/iotexproject/iotex-core/server/cronjob"
 )
 
 // Server is the iotex server instance containing all components.
@@ -39,6 +41,7 @@ type Server struct {
 	initializedSubChains map[uint32]bool
 	mutex                sync.RWMutex
 	subModuleCancel      context.CancelFunc
+	tasks                []lifecycle.StartStopper
 }
 
 // NewServer creates a new server
@@ -88,9 +91,17 @@ func newServer(cfg config.Config, testing bool) (*Server, error) {
 			return nil, errors.Wrap(err, "failed to add api server as subscriber")
 		}
 	}
+
+	tasks := []lifecycle.StartStopper{}
+	jobs := cronjob.NewCronJobs(cfg.Cronjob, p2pAgent, cs.Blockchain())
+	for i := range jobs {
+		tasks = append(tasks, routine.NewRecurringTask(jobs[i].Run, jobs[i].Interval()))
+	}
+
 	// TODO: explorer dependency deleted here at #1085, need to revive by migrating to api
 	chains[cs.ChainID()] = cs
 	dispatcher.AddSubscriber(cs.ChainID(), cs)
+	dispatcher.AddMsgSubscriber(apiServer)
 	svr := Server{
 		cfg:                  cfg,
 		p2pAgent:             p2pAgent,
@@ -99,6 +110,7 @@ func newServer(cfg config.Config, testing bool) (*Server, error) {
 		chainservices:        chains,
 		apiServers:           apiServers,
 		initializedSubChains: map[uint32]bool{},
+		tasks:                tasks,
 	}
 	// Setup sub-chain starter
 	// TODO: sub-chain infra should use main-chain API instead of protocol directly
@@ -125,13 +137,22 @@ func (s *Server) Start(ctx context.Context) error {
 	if err := s.dispatcher.Start(cctx); err != nil {
 		return errors.Wrap(err, "error when starting dispatcher")
 	}
-
+	for i := range s.tasks {
+		if err := s.tasks[i].Start(cctx); err != nil {
+			return errors.Wrapf(err, "error when starting task %v", i)
+		}
+	}
 	return nil
 }
 
 // Stop stops the server
 func (s *Server) Stop(ctx context.Context) error {
 	defer s.subModuleCancel()
+	for i := range s.tasks {
+		if err := s.tasks[i].Stop(ctx); err != nil {
+			return errors.Wrapf(err, "error when stopping task %v", i)
+		}
+	}
 	if err := s.p2pAgent.Stop(ctx); err != nil {
 		// notest
 		return errors.Wrap(err, "error when stopping P2P agent")