const (
BYTE = 1 << (10 * iota)
KILOBYTE
MEGABYTE
GIGABYTE
TERABYTE
)
const (
// DefaultSnapshotCatchUpEntries is the default number of entries
// to preserve in memory when a snapshot is taken. This is for
// slow followers to catch up.
DefaultSnapshotCatchUpEntries = uint64(20)
// DefaultSnapshotIntervalSize is the default snapshot interval. It is
// used if SnapshotIntervalSize is not provided in channel config options.
// It is needed to enforce snapshot being set.
DefaultSnapshotIntervalSize = 20 * MEGABYTE // 20 MB
// DefaultEvictionSuspicion is the threshold that a node will start
// suspecting its own eviction if it has been leaderless for this
// period of time.
DefaultEvictionSuspicion = time.Minute * 10
// DefaultLeaderlessCheckInterval is the interval that a chain checks
// its own leadership status.
DefaultLeaderlessCheckInterval = time.Second * 10
)
MaxSnapshotFiles defines max number of etcd/raft snapshot files to retain on filesystem. Snapshot files are read from newest to oldest, until first intact file is found. The more snapshot files we keep around, the more we mitigate the impact of a corrupted snapshots. This is exported for testing purpose. This MUST be greater equal than 1.
var MaxSnapshotFiles = 5
func CheckConfigMetadata(metadata *etcdraft.ConfigMetadata) error
CheckConfigMetadata validates Raft config metadata
func ConfChange(blockMetadata *etcdraft.BlockMetadata, confState *raftpb.ConfState) *raftpb.ConfChange
ConfChange computes Raft configuration changes based on current Raft configuration state and consenters IDs stored in RaftMetadata.
func ConfigChannelHeader(block *common.Block) (hdr *common.ChannelHeader, err error)
ConfigChannelHeader expects a config block and returns the header type of the config envelope wrapped in it, e.g. HeaderType_ORDERER_TRANSACTION
func ConfigEnvelopeFromBlock(block *common.Block) (*common.Envelope, error)
ConfigEnvelopeFromBlock extracts configuration envelope from the block based on the config type, i.e. HeaderType_ORDERER_TRANSACTION or HeaderType_CONFIG
func ConsensusMetadataFromConfigBlock(block *common.Block) (*etcdraft.ConfigMetadata, error)
ConsensusMetadataFromConfigBlock reads consensus metadata updates from the configuration block
func ConsentersToMap(consenters []*etcdraft.Consenter) map[string]struct{}
ConsentersToMap maps consenters into set where key is client TLS certificate
func EndpointconfigFromFromSupport(support consensus.ConsenterSupport) ([]cluster.EndpointCriteria, error)
EndpointconfigFromFromSupport extracts TLS CA certificates and endpoints from the ConsenterSupport
func ListSnapshots(logger *flogging.FabricLogger, snapDir string) []uint64
ListSnapshots returns a list of RaftIndex of snapshots stored on disk. If a file is corrupted, rename the file.
func MembershipByCert(consenters map[uint64]*etcdraft.Consenter) map[string]uint64
MembershipByCert convert consenters map into set encapsulated by map where key is client TLS certificate
func MetadataFromConfigUpdate(update *common.ConfigUpdate) (*etcdraft.ConfigMetadata, error)
MetadataFromConfigUpdate extracts consensus metadata from config update
func MetadataFromConfigValue(configValue *common.ConfigValue) (*etcdraft.ConfigMetadata, error)
MetadataFromConfigValue reads and translates configuration updates from config value into raft metadata
func MetadataHasDuplication(md *etcdraft.ConfigMetadata) error
MetadataHasDuplication returns an error if the metadata has duplication of consenters. A duplication is defined by having a server or a client TLS certificate that is found in two different consenters, regardless of the type of certificate (client/server).
func NodeExists(id uint64, nodes []uint64) bool
NodeExists returns trues if node id exists in the slice and false otherwise
func RaftPeers(consenterIDs []uint64) []raft.Peer
RaftPeers maps consenters to slice of raft.Peer
func ReadBlockMetadata(blockMetadata *common.Metadata, configMetadata *etcdraft.ConfigMetadata) (*etcdraft.BlockMetadata, error)
ReadBlockMetadata attempts to read raft metadata from block metadata, if available. otherwise, it reads raft metadata from config metadata supplied.
BlockPuller is used to pull blocks from other OSN
type BlockPuller interface {
PullBlock(seq uint64) *common.Block
HeightsByEndpoints() (map[string]uint64, error)
Close()
}
Chain implements consensus.Chain interface.
type Chain struct {
// this is exported so that test can use `Node.Status()` to get raft node status.
Node *node
Metrics *Metrics
// contains filtered or unexported fields
}
func NewChain(
support consensus.ConsenterSupport,
opts Options,
conf Configurator,
rpc RPC,
f CreateBlockPuller,
haltCallback func(),
observeC chan<- raft.SoftState) (*Chain, error)
NewChain constructs a chain object.
func (c *Chain) Configure(env *common.Envelope, configSeq uint64) error
Configure submits config type transactions for ordering.
func (c *Chain) Consensus(req *orderer.ConsensusRequest, sender uint64) error
Consensus passes the given ConsensusRequest message to the raft.Node instance
func (c *Chain) Errored() <-chan struct{}
Errored returns a channel that closes when the chain stops.
func (c *Chain) Halt()
Halt stops the chain.
func (c *Chain) Order(env *common.Envelope, configSeq uint64) error
Order submits normal type transactions for ordering.
func (c *Chain) Start()
Start instructs the orderer to begin serving the chain and keep it current.
func (c *Chain) Submit(req *orderer.SubmitRequest, sender uint64) error
Submit forwards the incoming request to: - the local serveRequest goroutine if this is leader - the actual leader via the transport mechanism The call fails if there's no leader elected yet.
func (c *Chain) WaitReady() error
WaitReady blocks when the chain: - is catching up with other nodes using snapshot
In any other case, it returns right away.
ChainGetter obtains instances of ChainSupport for the given channel
type ChainGetter interface {
// GetChain obtains the ChainSupport for the given channel.
// Returns nil, false when the ChainSupport for the given channel
// isn't found.
GetChain(chainID string) *multichannel.ChainSupport
}
Config contains etcdraft configurations
type Config struct {
WALDir string // WAL data of <my-channel> is stored in WALDir/<my-channel>
SnapDir string // Snapshots of <my-channel> are stored in SnapDir/<my-channel>
EvictionSuspicion string // Duration threshold that the node samples in order to suspect its eviction from the channel.
}
Configurator is used to configure the communication layer when the chain starts.
type Configurator interface {
Configure(channel string, newNodes []cluster.RemoteNode)
}
Consenter implements etcdraft consenter
type Consenter struct {
CreateChain func(chainName string)
InactiveChainRegistry InactiveChainRegistry
Dialer *cluster.PredicateDialer
Communication cluster.Communicator
*Dispatcher
Chains ChainGetter
Logger *flogging.FabricLogger
EtcdRaftConfig Config
OrdererConfig localconfig.TopLevel
Cert []byte
Metrics *Metrics
}
func New(
clusterDialer *cluster.PredicateDialer,
conf *localconfig.TopLevel,
srvConf comm.ServerConfig,
srv *comm.GRPCServer,
r *multichannel.Registrar,
icr InactiveChainRegistry,
metricsProvider metrics.Provider,
) *Consenter
New creates a etcdraft Consenter
func (c *Consenter) HandleChain(support consensus.ConsenterSupport, metadata *common.Metadata) (consensus.Chain, error)
HandleChain returns a new Chain instance or an error upon failure
func (c *Consenter) ReceiverByChain(channelID string) MessageReceiver
ReceiverByChain returns the MessageReceiver for the given channelID or nil if not found.
func (c *Consenter) TargetChannel(message proto.Message) string
TargetChannel extracts the channel from the given proto.Message. Returns an empty string on failure.
ConsenterCertificate denotes a TLS certificate of a consenter
type ConsenterCertificate []byte
func (conCert ConsenterCertificate) IsConsenterOfChannel(configBlock *common.Block) error
IsConsenterOfChannel returns whether the caller is a consenter of a channel by inspecting the given configuration block. It returns nil if true, else returns an error.
CreateBlockPuller is a function to create BlockPuller on demand. It is passed into chain initializer so that tests could mock this.
type CreateBlockPuller func() (BlockPuller, error)
CreateChainCallback creates a new chain
type CreateChainCallback func()
Dispatcher dispatches Submit and Step requests to the designated per chain instances
type Dispatcher struct {
Logger *flogging.FabricLogger
ChainSelector ReceiverGetter
}
func (d *Dispatcher) OnConsensus(channel string, sender uint64, request *orderer.ConsensusRequest) error
OnConsensus notifies the Dispatcher for a reception of a StepRequest from a given sender on a given channel
func (d *Dispatcher) OnSubmit(channel string, sender uint64, request *orderer.SubmitRequest) error
OnSubmit notifies the Dispatcher for a reception of a SubmitRequest from a given sender on a given channel
InactiveChainRegistry registers chains that are inactive
type InactiveChainRegistry interface {
// TrackChain tracks a chain with the given name, and calls the given callback
// when this chain should be created.
TrackChain(chainName string, genesisBlock *common.Block, createChain CreateChainCallback)
}
LedgerBlockPuller pulls blocks upon demand, or fetches them from the ledger.
type LedgerBlockPuller struct {
BlockPuller
BlockRetriever cluster.BlockRetriever
Height func() uint64
}
func (ledgerPuller *LedgerBlockPuller) PullBlock(seq uint64) *common.Block
MembershipChanges keeps information about membership changes introduced during configuration update
type MembershipChanges struct {
NewBlockMetadata *etcdraft.BlockMetadata
NewConsenters map[uint64]*etcdraft.Consenter
AddedNodes []*etcdraft.Consenter
RemovedNodes []*etcdraft.Consenter
ConfChange *raftpb.ConfChange
RotatedNode uint64
}
func ComputeMembershipChanges(oldMetadata *etcdraft.BlockMetadata, oldConsenters map[uint64]*etcdraft.Consenter, newConsenters []*etcdraft.Consenter) (mc *MembershipChanges, err error)
ComputeMembershipChanges computes membership update based on information about new conseters, returns two slices: a slice of added consenters and a slice of consenters to be removed
func (mc *MembershipChanges) Changed() bool
Changed indicates whether these changes actually do anything
func (mc *MembershipChanges) Rotated() bool
Rotated indicates whether the change was a rotation
func (mc *MembershipChanges) String() string
Stringer implements fmt.Stringer interface
MemoryStorage is currently backed by etcd/raft.MemoryStorage. This interface is defined to expose dependencies of fsm so that it may be swapped in the future. TODO(jay) Add other necessary methods to this interface once we need them in implementation, e.g. ApplySnapshot.
type MemoryStorage interface {
raft.Storage
Append(entries []raftpb.Entry) error
SetHardState(st raftpb.HardState) error
CreateSnapshot(i uint64, cs *raftpb.ConfState, data []byte) (raftpb.Snapshot, error)
Compact(compactIndex uint64) error
ApplySnapshot(snap raftpb.Snapshot) error
}
MessageReceiver receives messages
type MessageReceiver interface {
// Consensus passes the given ConsensusRequest message to the MessageReceiver
Consensus(req *orderer.ConsensusRequest, sender uint64) error
// Submit passes the given SubmitRequest message to the MessageReceiver
Submit(req *orderer.SubmitRequest, sender uint64) error
}
type Metrics struct {
ClusterSize metrics.Gauge
IsLeader metrics.Gauge
CommittedBlockNumber metrics.Gauge
SnapshotBlockNumber metrics.Gauge
LeaderChanges metrics.Counter
ProposalFailures metrics.Counter
DataPersistDuration metrics.Histogram
NormalProposalsReceived metrics.Counter
ConfigProposalsReceived metrics.Counter
}
func NewMetrics(p metrics.Provider) *Metrics
Options contains all the configurations relevant to the chain.
type Options struct {
RaftID uint64
Clock clock.Clock
WALDir string
SnapDir string
SnapshotIntervalSize uint32
// This is configurable mainly for testing purpose. Users are not
// expected to alter this. Instead, DefaultSnapshotCatchUpEntries is used.
SnapshotCatchUpEntries uint64
MemoryStorage MemoryStorage
Logger *flogging.FabricLogger
TickInterval time.Duration
ElectionTick int
HeartbeatTick int
MaxSizePerMsg uint64
MaxInflightBlocks int
// BlockMetdata and Consenters should only be modified while under lock
// of raftMetadataLock
BlockMetadata *etcdraft.BlockMetadata
Consenters map[uint64]*etcdraft.Consenter
// MigrationInit is set when the node starts right after consensus-type migration
MigrationInit bool
Metrics *Metrics
Cert []byte
EvictionSuspicion time.Duration
LeaderCheckInterval time.Duration
}
PeriodicCheck checks periodically a condition, and reports the cumulative consecutive period the condition was fulfilled.
type PeriodicCheck struct {
Logger *flogging.FabricLogger
CheckInterval time.Duration
Condition func() bool
Report func(cumulativePeriod time.Duration)
// contains filtered or unexported fields
}
func (pc *PeriodicCheck) Run()
Run runs the PeriodicCheck
func (pc *PeriodicCheck) Stop()
Stop stops the periodic checks
RPC is used to mock the transport layer in tests.
type RPC interface {
SendConsensus(dest uint64, msg *orderer.ConsensusRequest) error
SendSubmit(dest uint64, request *orderer.SubmitRequest) error
}
RaftStorage encapsulates storages needed for etcd/raft data, i.e. memory, wal
type RaftStorage struct {
SnapshotCatchUpEntries uint64
// contains filtered or unexported fields
}
func CreateStorage(
lg *flogging.FabricLogger,
walDir string,
snapDir string,
ram MemoryStorage,
) (*RaftStorage, error)
CreateStorage attempts to create a storage to persist etcd/raft data. If data presents in specified disk, they are loaded to reconstruct storage state.
func (rs *RaftStorage) ApplySnapshot(snap raftpb.Snapshot)
ApplySnapshot applies snapshot to local memory storage
func (rs *RaftStorage) Close() error
Close closes storage
func (rs *RaftStorage) Snapshot() raftpb.Snapshot
Snapshot returns the latest snapshot stored in memory
func (rs *RaftStorage) Store(entries []raftpb.Entry, hardstate raftpb.HardState, snapshot raftpb.Snapshot) error
Store persists etcd/raft data
func (rs *RaftStorage) TakeSnapshot(i uint64, cs raftpb.ConfState, data []byte) error
TakeSnapshot takes a snapshot at index i from MemoryStorage, and persists it to wal and disk.
ReceiverGetter obtains instances of MessageReceiver given a channel ID
type ReceiverGetter interface {
// ReceiverByChain returns the MessageReceiver if it exists, or nil if it doesn't
ReceiverByChain(channelID string) MessageReceiver
}