node_manager

README

StreamingFast Node Manager

This service is a wrapping process that operates blockchain nodes as part of StreamingFast.

Installation

Build and run from here:

• EOSIO on StreamingFast
• Ethereum on StreamingFast

Overview

1. Operator (process commands, owns the superviser, owns the http handlers that sends the commands to him)

2. Superviser is the the one managing the actual blockchain software (nodeos, geth..). It is made of an embedded generic superviser struct, plus specific nodeos/geth embedding it. It owns the plugins.

Shutdown pattern for Node-Manager only:

App creates:

• Superviser
• Operator (+superviser)

So, the ownership is app -> operator -> superviser

• app.OnTerminating(operator.Shutdown())
• operator.OnTerminating(sendCmd:"maintenance", superviser.Shutdown())
• superviser.OnTerminating(superviser.Stop() (blocking))

Shutdown pattern for Mindreader:

App creates:

• Superviser
• Operator (+superviser)
• mindreaderPlugin (has call back to set maintenance on operator and stopBlockReached)

App sets:

• superviser.RegisterLogPlugin(mindreaderPlugin)

So, the ownership is app -> operator -> superviser -> mindreader

• app.OnTerminating(operator.Shutdown())
• operator.OnTerminating(sendCmd:"maintenance", superviser.Shutdown())
• superviser.OnTerminating(mindreader.Shutdown(), then endLogPlugins)
• superviser.OnTerminated(endLogPlugins)
• mindreader.OnTerminating(async operator.Shutdown(), wait consumeFlowDone)
  • mindreader::archiver closes consumeFlowDone when superviser.endLogPlugins(+upload completed)
  • mindreader shuts itself down when stopBlockNum reached
• mindreader.OnTerminated -> app.Shutdown()

Contributing

Issues and PR in this repo related strictly to the core manageos engine

Report any protocol-specific issues in their respective repositories

Please first refer to the general StreamingFast contribution guide, if you wish to contribute to this code base.

This codebase uses unit tests extensively, please write and run tests.

License

Apache 2.0

Documentation

Index

• Variables
• func AugmentStackSizeLimit() error
• type ChainSuperviser
• type DeepMindDebuggable
• type HeadBlockUpdater
• type MetricsAndReadinessManager
  • func NewMetricsAndReadinessManager(headBlockTimeDrift *dmetrics.HeadTimeDrift, ...) *MetricsAndReadinessManager
  • func (m *MetricsAndReadinessManager) IsReady() bool
  • func (m *MetricsAndReadinessManager) Launch()
  • func (m *MetricsAndReadinessManager) UpdateHeadBlock(block *bstream.Block) error
• type MonitorableChainSuperviser
• type OnBlockWritten
• type ProducerChainSuperviser
• type ProductionEvent
• type ProductionState
  • func (s ProductionState) String() string
• type Readiness
• type ShutterInterface
• type StartOption

Variables

var DisableDebugDeepmindOption = StartOption("disable-debug-firehose-logs")

var EnableDebugDeepmindOption = StartOption("enable-debug-firehose-logs")

Functions

func AugmentStackSizeLimit

func AugmentStackSizeLimit() error

Types

type ChainSuperviser

type ChainSuperviser interface {
	ShutterInterface

	GetCommand() string
	GetName() string
	ServerID() (string, error)

	RegisterLogPlugin(plugin logplugin.LogPlugin)
	Start(options ...StartOption) error
	Stop() error

	IsRunning() bool
	Stopped() <-chan struct{}

	LastExitCode() int
	LastLogLines() []string
	LastSeenBlockNum() uint64
}

type DeepMindDebuggable

type DeepMindDebuggable interface {
	DebugDeepMind(enabled bool)
}

type HeadBlockUpdater

type HeadBlockUpdater func(block *bstream.Block) error

type MetricsAndReadinessManager

type MetricsAndReadinessManager struct {
	// contains filtered or unexported fields
}

func NewMetricsAndReadinessManager

func NewMetricsAndReadinessManager(headBlockTimeDrift *dmetrics.HeadTimeDrift, headBlockNumber *dmetrics.HeadBlockNum, appReadiness *dmetrics.AppReadiness, readinessMaxLatency time.Duration) *MetricsAndReadinessManager

func (*MetricsAndReadinessManager) IsReady

func (m *MetricsAndReadinessManager) IsReady() bool

func (*MetricsAndReadinessManager) Launch

func (m *MetricsAndReadinessManager) Launch()

func (*MetricsAndReadinessManager) UpdateHeadBlock

func (m *MetricsAndReadinessManager) UpdateHeadBlock(block *bstream.Block) error

type MonitorableChainSuperviser

type MonitorableChainSuperviser interface {
	Monitor()
}

type OnBlockWritten

type OnBlockWritten func(block *bstream.Block) error

type ProducerChainSuperviser

type ProducerChainSuperviser interface {
	IsProducing() (bool, error)
	IsActiveProducer() bool

	ResumeProduction() error
	PauseProduction() error

	WaitUntilEndOfNextProductionRound(timeout time.Duration) error
}

type ProductionEvent

type ProductionEvent int

const (
	EventProduced ProductionEvent = iota
	EventReceived
)

type ProductionState

type ProductionState int

const (
	StatePre       ProductionState = iota // Just before we produce, don't restart
	StateProducing                        // We're producing right now
	StatePost                             // Right after production
	StateStale                            // We haven't produced for 12 minutes
)

func (ProductionState) String

func (s ProductionState) String() string

type Readiness

type Readiness interface {
	IsReady() bool
}

type ShutterInterface

type ShutterInterface interface {
	Shutdown(error)
	OnTerminating(func(error))
	OnTerminated(func(error))
	IsTerminated() bool
	IsTerminating() bool
	Terminated() <-chan struct{}
}

type StartOption

type StartOption string