dmesh

README

dfuse Mesh Library

This repository contains all common code around service discovery across our various services. It is part of dfuse.

Usage

See example usage in search.

Contributing

Issues and PR in this repo related strictly to the dmesh library.

Report any protocol-specific issues in their respective repositories

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

License

Apache 2.0

Documentation

Index

• Constants
• Variables
• func GetLocalIP() string
• func Observe(ctx context.Context, store *clientv3.Client, ...) <-chan *PeerEvent
• func ValidateServiceList(services []string) ([]string, error)
• type BlockRangeData
  • func NewTestBlockRangeData(tail, irr, head uint64) BlockRangeData
  • func (d BlockRangeData) Blocks() (tail, irr, head uint64)
  • func (d BlockRangeData) HeadBlockPointers() (uint64, string, uint64, string, uint64, string)
• type ConnectablePeer
• type GenericPeer
  • func NewTestGenericPeer(serviceVersion, serviceName, addr string) GenericPeer
  • func NewTestReadyGenericPeer(serviceVersion, serviceName, addr string) GenericPeer
  • func (p *GenericPeer) Addr() string
  • func (p *GenericPeer) Close() error
  • func (p *GenericPeer) Conn() *grpc.ClientConn
  • func (p *GenericPeer) Key() string
  • func (p *GenericPeer) ResetConn() error
  • func (p *GenericPeer) SetConn(grpcConn *grpc.ClientConn)
  • func (p *GenericPeer) SetKey(key string) error
• type HeadBlockData
• type MergeablePeer
• type Peer
  • func UnmarshalPeer(key string, peerData []byte) (Peer, error)
• type PeerEvent
• type RelayerPeer
  • func NewRelayerPeer(serviceVersion, listenAddr string, highBlockNum uint64) *RelayerPeer
  • func (p *RelayerPeer) Locked(f func()) Peer
  • func (p *RelayerPeer) Merge(in Peer)
• type SearchPeer
  • func NewSearchArchivePeer(serviceVersion, listenAddr string, hasMovingTail, hasMovingHead bool, ...) *SearchPeer
  • func NewSearchForkResolverPeer(serviceVersion, listenAddr string, pollingDuration time.Duration) *SearchPeer
  • func NewSearchHeadPeer(serviceVersion, listenAddr string, shardSize uint64, tierLevel uint32, ...) *SearchPeer
  • func (p *SearchPeer) Locked(f func()) Peer
  • func (p *SearchPeer) Merge(in Peer)
  • func (p *SearchPeer) VirtualHead() uint64

Constants

const (
	EventUnkown = "unknown"
	EventSync   = "sync"
	EventUpdate = "update"
	EventDelete = "delete"
)

const (
	SearchServiceName  = "search"
	RelayerServiceName = "relayer"
)

const RelayerPeerPollingDuration = (10 * 365 * 24 * time.Hour)

const RelayerPeerType = "relayer"

Variables

var Hostname, _ = os.Hostname()

Functions

func GetLocalIP

func GetLocalIP() string

func Observe

func Observe(ctx context.Context, store *clientv3.Client, namespacePrefix, peerPrefix string) <-chan *PeerEvent

func ValidateServiceList

func ValidateServiceList(services []string) ([]string, error)

Cleans up service prefixes to watch

/v2/service-name-123/ -> /v2/service-name-123/ /v2/service-name-123 -> /v2/service-name-123/ v2/service-name-123/ -> /v2/service-name-123/ v2/service-name-123 -> /v2/service-name-123/

Types

type BlockRangeData

type BlockRangeData struct {
	TailBlock     uint64     `json:"tailBlockNum,omitempty"`
	TailBlockID   string     `json:"tailBlockID,omitempty"`
	TailBlockTime *time.Time `json:"tailBlockTime,omitempty"`
	HeadBlockData
}

func NewTestBlockRangeData

func NewTestBlockRangeData(tail, irr, head uint64) BlockRangeData

func (BlockRangeData) Blocks

func (d BlockRangeData) Blocks() (tail, irr, head uint64)

func (BlockRangeData) HeadBlockPointers

func (d BlockRangeData) HeadBlockPointers() (uint64, string, uint64, string, uint64, string)

type ConnectablePeer

type ConnectablePeer interface {
	Addr() string
	Conn() *grpc.ClientConn
	SetConn(conn *grpc.ClientConn)
	Close() error
	ResetConn() error
}

type GenericPeer

type GenericPeer struct {
	sync.RWMutex

	Ready bool       `json:"ready,omitempty"` // should be ready when service is live
	Host  string     `json:"host,omitempty"`  // search-liverouter-v2-1-59f98966b8-vzzsb
	Boot  *time.Time `json:"boot,omitempty"`
	// contains filtered or unexported fields
}

func NewTestGenericPeer

func NewTestGenericPeer(serviceVersion, serviceName, addr string) GenericPeer

func NewTestReadyGenericPeer

func NewTestReadyGenericPeer(serviceVersion, serviceName, addr string) GenericPeer

func (*GenericPeer) Addr

func (p *GenericPeer) Addr() string

ConnectiblePeer interface

func (*GenericPeer) Close

func (p *GenericPeer) Close() error

func (*GenericPeer) Conn

func (p *GenericPeer) Conn() *grpc.ClientConn

func (*GenericPeer) Key

func (p *GenericPeer) Key() string

Peer interface

func (*GenericPeer) ResetConn

func (p *GenericPeer) ResetConn() error

func (*GenericPeer) SetConn

func (p *GenericPeer) SetConn(grpcConn *grpc.ClientConn)

func (*GenericPeer) SetKey

func (p *GenericPeer) SetKey(key string) error

type HeadBlockData

type HeadBlockData struct {
	IrrBlock      uint64     `json:"irrBlockNum,omitempty"`
	IrrBlockID    string     `json:"irrBlockID,omitempty"`
	IrrBlockTime  *time.Time `json:"irrBlockTime,omitempty"`
	HeadBlock     uint64     `json:"headBlockNum,omitempty"`
	HeadBlockID   string     `json:"headBlockID,omitempty"`
	HeadBlockTime *time.Time `json:"headBlockTime,omitempty"`
}

type MergeablePeer

type MergeablePeer interface {
	// Merge copies from `in` any fields that could have moved, when
	// watching from etcd.
	Merge(in Peer)
}

type Peer

type Peer interface {
	SetKey(string) error
	Key() string
}

func UnmarshalPeer

func UnmarshalPeer(key string, peerData []byte) (Peer, error)

type PeerEvent

type PeerEvent struct {
	EventName string
	Peer      Peer
	PeerKey   string
}

type RelayerPeer

type RelayerPeer struct {
	GenericPeer
	HeadBlockData

	HighBlockNum uint64 `json:"high_blk"`
}

func NewRelayerPeer

func NewRelayerPeer(serviceVersion, listenAddr string, highBlockNum uint64) *RelayerPeer

func (*RelayerPeer) Locked

func (p *RelayerPeer) Locked(f func()) Peer

func (*RelayerPeer) Merge

func (p *RelayerPeer) Merge(in Peer)

type SearchPeer

type SearchPeer struct {
	GenericPeer
	BlockRangeData

	ServesResolveForks bool `json:"resolveForks,omitempty"` // peer can resolve forks
	ServesReversible   bool `json:"reversible,omitempty"`   // peer serves the reversible segment of the chain, and is capable of navigating it
	HasMovingHead      bool `json:"headMoves,omitempty"`    // false for frozen ranges
	HasMovingTail      bool `json:"tailMoves,omitempty"`    // whether the process truncates its lower blocks

	ShardSize uint64 `json:"shardSize"`
	TierLevel uint32 `json:"tier"` // Frozen archives, first segments, are lower tiers (0, 1), and those at the tip are higher, more chances we roll out new ones at the tip of the chain.
	// contains filtered or unexported fields
}

func NewSearchArchivePeer

func NewSearchArchivePeer(serviceVersion, listenAddr string, hasMovingTail, hasMovingHead bool, shardSize uint64, tierLevel uint32, pollingDuration time.Duration) *SearchPeer

func NewSearchForkResolverPeer

func NewSearchForkResolverPeer(serviceVersion, listenAddr string, pollingDuration time.Duration) *SearchPeer

func NewSearchHeadPeer

func NewSearchHeadPeer(serviceVersion, listenAddr string, shardSize uint64, tierLevel uint32, pollingDuration time.Duration) *SearchPeer

func (*SearchPeer) Locked

func (p *SearchPeer) Locked(f func()) Peer

func (*SearchPeer) Merge

func (p *SearchPeer) Merge(in Peer)

func (*SearchPeer) VirtualHead

func (p *SearchPeer) VirtualHead() uint64