cluster

Documentation

Index

• Variables
• type Engine
  • func NewEngine(name string, disco discovery.Provider, host *discovery.Node, opts ...Option) (*Engine, error)
  • func (x *Engine) Actors(ctx context.Context, timeout time.Duration) ([]*internalpb.ActorRef, error)
  • func (x *Engine) Events() <-chan *Event
  • func (x *Engine) GetActor(ctx context.Context, actorName string) (*internalpb.ActorRef, error)
  • func (x *Engine) GetPartition(actorName string) int
  • func (x *Engine) GetState(ctx context.Context, peerAddress string) (*internalpb.PeerState, error)
  • func (x *Engine) IsLeader(ctx context.Context) bool
  • func (x *Engine) IsRunning() bool
  • func (x *Engine) Peers(ctx context.Context) ([]*Peer, error)
  • func (x *Engine) PutActor(ctx context.Context, actor *internalpb.ActorRef) error
  • func (x *Engine) RemoveActor(ctx context.Context, actorName string) error
  • func (x *Engine) SchedulerJobKeyExists(ctx context.Context, key string) (bool, error)
  • func (x *Engine) SetSchedulerJobKey(ctx context.Context, key string) error
  • func (x *Engine) Start(ctx context.Context) error
  • func (x *Engine) Stop(ctx context.Context) error
  • func (x *Engine) UnsetSchedulerJobKey(ctx context.Context, key string) error
• type Event
• type EventType
  • func (x EventType) String() string
• type Interface
• type Option
  • func WithHasher(hasher hash.Hasher) Option
  • func WithKVStoreSize(size uint64) Option
  • func WithLogger(logger log.Logger) Option
  • func WithMinimumPeersQuorum(count uint32) Option
  • func WithPartitionsCount(count uint64) Option
  • func WithReadQuorum(count uint32) Option
  • func WithReadTimeout(timeout time.Duration) Option
  • func WithReplicaCount(count uint32) Option
  • func WithShutdownTimeout(timeout time.Duration) Option
  • func WithTLS(serverConfig, clientConfig *tls.Config) Option
  • func WithWriteQuorum(count uint32) Option
  • func WithWriteTimeout(timeout time.Duration) Option
• type OptionFunc
  • func (f OptionFunc) Apply(c *Engine)
• type Peer
  • func (peer Peer) PeerAddress() string

Variables

var (
	// ErrActorNotFound is return when an actor is not found
	ErrActorNotFound = errors.New("actor not found")
	// ErrPeerSyncNotFound is returned when a peerSync record is not found
	ErrPeerSyncNotFound = errors.New("peerSync record not found")
	// ErrInvalidTLSConfiguration is returned whent the TLS configuration is not properly set
	ErrInvalidTLSConfiguration = errors.New("TLS configuration is invalid")
	// ErrEngineNotRunning is returned when the cluster engine is not running
	ErrEngineNotRunning = errors.New("engine is not running")
)

Types

type Engine

type Engine struct {
	*sync.Mutex
	// contains filtered or unexported fields
}

Engine represents the Engine

func NewEngine

func NewEngine(name string, disco discovery.Provider, host *discovery.Node, opts ...Option) (*Engine, error)

NewEngine creates an instance of cluster Engine

func (*Engine) Actors

func (x *Engine) Actors(ctx context.Context, timeout time.Duration) ([]*internalpb.ActorRef, error)

Actors returns all actors in the cluster at any given time

func (*Engine) Events

func (x *Engine) Events() <-chan *Event

Events returns a channel where cluster events are published

func (*Engine) GetActor

func (x *Engine) GetActor(ctx context.Context, actorName string) (*internalpb.ActorRef, error)

GetActor fetches an actor from the Node

func (*Engine) GetPartition

func (x *Engine) GetPartition(actorName string) int

GetPartition returns the partition where a given actor is stored

func (*Engine) GetState

func (x *Engine) GetState(ctx context.Context, peerAddress string) (*internalpb.PeerState, error)

GetState fetches a given peer state

func (*Engine) IsLeader

func (x *Engine) IsLeader(ctx context.Context) bool

IsLeader states whether the given cluster node is a leader or not at a given point in time in the cluster

func (*Engine) IsRunning

func (x *Engine) IsRunning() bool

IsRunning returns true when the cluster engine is running

func (*Engine) Peers

func (x *Engine) Peers(ctx context.Context) ([]*Peer, error)

Peers returns a channel containing the list of peers at a given time

func (*Engine) PutActor

func (x *Engine) PutActor(ctx context.Context, actor *internalpb.ActorRef) error

PutActor pushes to the cluster the peer sync request

func (*Engine) RemoveActor

func (x *Engine) RemoveActor(ctx context.Context, actorName string) error

RemoveActor removes a given actor from the cluster. An actor is removed from the cluster when this actor has been passivated.

func (*Engine) SchedulerJobKeyExists

func (x *Engine) SchedulerJobKeyExists(ctx context.Context, key string) (bool, error)

SchedulerJobKeyExists checks the existence of a given key

func (*Engine) SetSchedulerJobKey

func (x *Engine) SetSchedulerJobKey(ctx context.Context, key string) error

SetSchedulerJobKey sets a given key to the cluster

func (*Engine) Start

func (x *Engine) Start(ctx context.Context) error

Start starts the Engine.

func (*Engine) Stop

func (x *Engine) Stop(ctx context.Context) error

Stop stops the Engine gracefully

func (*Engine) UnsetSchedulerJobKey

func (x *Engine) UnsetSchedulerJobKey(ctx context.Context, key string) error

UnsetSchedulerJobKey unsets the already set given key in the cluster

type Event

type Event struct {
	Payload *anypb.Any
	Type    EventType
}

Event defines the cluster event

type EventType

type EventType int

const (
	NodeJoined EventType = iota
	NodeLeft
)

func (EventType) String

func (x EventType) String() string

type Interface

type Interface interface {
	// Start starts the cluster engine
	Start(ctx context.Context) error
	// Stop stops the cluster engine
	Stop(ctx context.Context) error
	// PutActor replicates onto the Node the metadata of an actor
	PutActor(ctx context.Context, actor *internalpb.ActorRef) error
	// GetActor fetches an actor from the Node
	GetActor(ctx context.Context, actorName string) (*internalpb.ActorRef, error)
	// GetPartition returns the partition where a given actor is stored
	GetPartition(actorName string) int
	// SetSchedulerJobKey sets a given key to the cluster
	SetSchedulerJobKey(ctx context.Context, key string) error
	// SchedulerJobKeyExists checks the existence of a given key
	SchedulerJobKeyExists(ctx context.Context, key string) (bool, error)
	// UnsetSchedulerJobKey unsets the already set given key in the cluster
	UnsetSchedulerJobKey(ctx context.Context, key string) error
	// RemoveActor removes a given actor from the cluster.
	// An actor is removed from the cluster when this actor has been passivated.
	RemoveActor(ctx context.Context, actorName string) error
	// Events returns a channel where cluster events are published
	Events() <-chan *Event
	// Peers returns a channel containing the list of peers at a given time
	Peers(ctx context.Context) ([]*Peer, error)
	// GetState fetches a given peer state
	GetState(ctx context.Context, peerAddress string) (*internalpb.PeerState, error)
	// IsLeader states whether the given cluster node is a leader or not at a given
	// point in time in the cluster
	IsLeader(ctx context.Context) bool
	// Actors returns all actors in the cluster at any given time
	Actors(ctx context.Context, timeout time.Duration) ([]*internalpb.ActorRef, error)
	// IsRunning returns true when the cluster engine is running
	IsRunning() bool
}

Interface defines the Node interface

type Option

type Option interface {
	// Apply sets the Option value of a config.
	Apply(cl *Engine)
}

Option is the interface that applies a configuration option.

func WithHasher

func WithHasher(hasher hash.Hasher) Option

WithHasher sets the custom hasher

func WithKVStoreSize

func WithKVStoreSize(size uint64) Option

WithKVStoreSize sets the cluster table storage size

func WithLogger

func WithLogger(logger log.Logger) Option

WithLogger sets the logger

func WithMinimumPeersQuorum

func WithMinimumPeersQuorum(count uint32) Option

WithMinimumPeersQuorum sets the minimum number of nodes to form a quorum

func WithPartitionsCount

func WithPartitionsCount(count uint64) Option

WithPartitionsCount sets the total number of partitions

func WithReadQuorum

func WithReadQuorum(count uint32) Option

WithReadQuorum sets the read quorum

func WithReadTimeout

func WithReadTimeout(timeout time.Duration) Option

WithReadTimeout sets the Node read timeout. This timeout specifies the timeout of a data retrieval

func WithReplicaCount

func WithReplicaCount(count uint32) Option

WithReplicaCount sets replica count

func WithShutdownTimeout

func WithShutdownTimeout(timeout time.Duration) Option

WithShutdownTimeout sets the Node shutdown timeout.

func WithTLS

func WithTLS(serverConfig, clientConfig *tls.Config) Option

WithTLS sets the various TLS for both Server and Client configuration. Bear in mind that both Client and Server must have the same root CA for successful handshake and authentication

func WithWriteQuorum

func WithWriteQuorum(count uint32) Option

WithWriteQuorum sets the write quorum

func WithWriteTimeout

func WithWriteTimeout(timeout time.Duration) Option

WithWriteTimeout sets the Node write timeout. This timeout specifies the timeout of a data replication

type OptionFunc

type OptionFunc func(eng *Engine)

OptionFunc implements the Option interface.

func (OptionFunc) Apply

func (f OptionFunc) Apply(c *Engine)

Apply applies the Node's option

type Peer

type Peer struct {
	// Host represents the peer address.
	Host string
	// PeersPort represents the peer port
	PeersPort int
	// Coordinator states that the given peer is the leader not.
	// A peer is a coordinator when it is the oldest node in the cluster
	Coordinator bool
}

Peer defines the peer info

func (Peer) PeerAddress

func (peer Peer) PeerAddress() string

PeerAddress returns address the node's peers will use to connect to