store

Documentation

Overview

* Package store provides an implementation for a accessing a Raft * backed job state machine store

Index

• Variables
• func JobToJobProto(j state.Job) *v1.JobProto
• func NewJobFromJobProto(jp *v1.JobProto) state.Job
• func NewSnapshotFrom(jsm state.JSM) (*snapshot, error)
• func RestoreSnapshotTo(rdr io.Reader, jsm state.JSM, timeout time.Duration) error
• func ToStrings(clientIds []state.ClientID) ([]string, error)
• type ClientURI
  • func NewClientURI(proxyID string, clientID string) *ClientURI
  • func ParseClientURI(clientID state.ClientID) (*ClientURI, error)
  • func (c *ClientURI) ToClientID() state.ClientID
  • func (c *ClientURI) Validate() error
• type Config
• type Store
  • func NewStore(c *Config) (*Store, error)
  • func (s *Store) ApplyOp(req *v1.ApplyOpRequest) *v1.ApplyOpResponse
  • func (s *Store) BootstrapCluster(nc map[string]string) error
  • func (s *Store) Close() error
  • func (s *Store) GetRaftConfiguration() (*raft.Configuration, error)
  • func (s *Store) IsLeader() bool
  • func (s *Store) Join(nodeID, addr string) error
  • func (s *Store) Leave(nodeID string) error
  • func (s *Store) NowSeconds() int64
  • func (s *Store) Open() error
  • func (s *Store) Ready() bool
  • func (s *Store) Snapshot() error
  • func (s *Store) TransferLeadership() error

Variables

var (
	ErrInvalidURI       = errors.New("invalid uri")
	ErrValidationFailed = errors.New("uri validation failed")
	ErrInvalidSchema    = errors.New("uri schema did not match or not present")
)

var (
	// ErrRaftConfig is returned when an error is encountered retrieving
	// the raft configuration.
	ErrRaftConfig = errors.New("raft configuration error")

	// ErrNodeNotLeader is returned, when the request requires the current
	// node to be a leader to execute, but is a not a raft leader.
	ErrNotRaftLeader = errors.New("node is not a raft leader")

	// ErrNodeNotFound is returned. when the specified node is not found in raft configuration
	ErrNodeNotFound = errors.New("node is not found in raft configuration")
)

Functions

func JobToJobProto

func JobToJobProto(j state.Job) *v1.JobProto

func NewJobFromJobProto

func NewJobFromJobProto(jp *v1.JobProto) state.Job

func NewSnapshotFrom

func NewSnapshotFrom(jsm state.JSM) (*snapshot, error)

func RestoreSnapshotTo

func RestoreSnapshotTo(rdr io.Reader, jsm state.JSM, timeout time.Duration) error

func ToStrings

func ToStrings(clientIds []state.ClientID) ([]string, error)

Types

type ClientURI

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

func NewClientURI

func NewClientURI(proxyID string, clientID string) *ClientURI

func ParseClientURI

func ParseClientURI(clientID state.ClientID) (*ClientURI, error)

func (*ClientURI) ToClientID

func (c *ClientURI) ToClientID() state.ClientID

func (*ClientURI) Validate

func (c *ClientURI) Validate() error

type Config

type Config struct {
	// retainSnapshotCount indicates the max, number of snapshots to retain
	RetainSnasphotCount int

	// The MaxPool controls how many connections we will pool. The
	MaxPool int

	// SnapshotThreshold controls how many outstanding logs there must be before
	// we perform a snapshot. This is to prevent excessive snapshots when we can
	// just replay a small set of logs.
	SnapshotThreshold uint64

	// TrailingLogs controls how many logs we leave after a snapshot. This is
	// used so that we can quickly replay logs on a follower instead of being
	// forced to send an entire snapshot.
	TrailingLogs uint64

	// SnapshotInterval controls how often we check if we should perform a snapshot.
	// We randomly stagger between this value and 2x this value to avoid the entire
	// cluster from performing a snapshot at once.
	SnapshotInterval time.Duration

	// RaftTimeout is the max. duration for a raft apply op.
	// timeout is used to apply I/O deadlines. For InstallSnapshot, we multiply
	// the timeout by (SnapshotSize / TimeoutScale).
	RaftTimeout time.Duration

	// RestoreTimeout is the max. duration for a restore operation
	RestoreTimeout time.Duration

	// RootDir is the root directory where store data is persisted
	RootDir string

	// RaftBindAddr is the bind address for raft tcp conn.
	RaftBindAddr string

	// Inmem is a boolean, controls if the data is persisted
	Inmem bool

	// When set to false skips fsync calls to the bolt store
	// it is not recommended to use NoSync as false on boltDb
	// Refer: https://github.com/boltdb/bolt/issues/612
	LogNoSync bool

	// local node id of this node
	LocalNodeID string
}

type Store

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

func NewStore

func NewStore(c *Config) (*Store, error)

func (*Store) ApplyOp

func (s *Store) ApplyOp(req *v1.ApplyOpRequest) *v1.ApplyOpResponse

func (*Store) BootstrapCluster

func (s *Store) BootstrapCluster(nc map[string]string) error

BootstrapCluster attempts to do a one-time bootstrap of the cluster the input is a map of nodeID & corresponding raft address entries

func (*Store) Close

func (s *Store) Close() error

func (*Store) GetRaftConfiguration

func (s *Store) GetRaftConfiguration() (*raft.Configuration, error)

func (*Store) IsLeader

func (s *Store) IsLeader() bool

func (*Store) Join

func (s *Store) Join(nodeID, addr string) error

Join joins a node, identified by nodeID and located at addr, to this store. The node must be ready to respond to Raft communications at that address.

It is required that the node that this is called into is a leader node.

func (*Store) Leave

func (s *Store) Leave(nodeID string) error

Leave, allows a node (specified by nodeID_ to leave the cluster.

It is required that the node that this is called into is a leader node.

func (*Store) NowSeconds

func (s *Store) NowSeconds() int64

func (*Store) Open

func (s *Store) Open() error

Open opens the store. If enableSingle is set, and there are no existing peers, then this node becomes the first node, and therefore leader, of the cluster. localID should be the server identifier for this node.

func (*Store) Ready

func (s *Store) Ready() bool

func (*Store) Snapshot

func (s *Store) Snapshot() error

func (*Store) TransferLeadership

func (s *Store) TransferLeadership() error