server

Documentation

Index

• Constants
• Variables
• func GetLogLevel(level string) (uint32, error)
• type ClusteringConfig
• type CommitLog
• type Config
  • func NewConfig(configFile string) (*Config, error)
  • func NewDefaultConfig() *Config
• type LogConfig
• type Server
  • func New(config *Config) *Server
  • func RunServerWithConfig(config *Config) (*Server, error)
  • func (s *Server) Apply(l *raft.Log) interface{}
  • func (s *Server) Restore(snapshot io.ReadCloser) error
  • func (s *Server) Snapshot() (raft.FSMSnapshot, error)
  • func (s *Server) Start() error
  • func (s *Server) Stop() error

Constants

const (
	// DefaultNamespace is the default cluster namespace to use if one is not
	// specified.
	DefaultNamespace = "liftbridge-default"

	// DefaultPort is the port to bind to if one is not specified.
	DefaultPort = 9292
)

Variables

var ErrStreamExists = errors.New("stream already exists")

ErrStreamExists is returned by CreateStream when attempting to create a stream that already has the provided subject and name.

Functions

func GetLogLevel

func GetLogLevel(level string) (uint32, error)

GetLogLevel converts the level string to its corresponding int value. It returns an error if the level is invalid.

Types

type ClusteringConfig

type ClusteringConfig struct {
	ServerID                string
	Namespace               string
	RaftSnapshots           int
	RaftSnapshotThreshold   uint64
	RaftCacheSize           int
	RaftBootstrap           bool
	RaftBootstrapPeers      []string
	RaftLogging             bool
	ReplicaMaxLagTime       time.Duration
	ReplicaMaxLeaderTimeout time.Duration
	ReplicaFetchTimeout     time.Duration
}

ClusteringConfig contains settings for controlling cluster behavior.

type CommitLog

type CommitLog interface {
	// Delete closes the log and removes all data associated with it from the
	// filesystem.
	Delete() error

	// NewReaderUncommitted returns an io.Reader which reads data from the log
	// starting at the given offset.
	NewReaderUncommitted(ctx context.Context, offset int64) (io.Reader, error)

	// NewReaderCommitted returns an io.Reader which reads only committed data
	// from the log starting at the given offset.
	NewReaderCommitted(ctx context.Context, offset int64) (io.Reader, error)

	// Truncate removes all messages from the log starting at the given offset.
	Truncate(offset int64) error

	// NewestOffset returns the offset of the last message in the log.
	NewestOffset() int64

	// OldestOffset returns the offset of the first message in the log.
	OldestOffset() int64

	// SetHighWatermark sets the high watermark on the log. All messages up to
	// and including the high watermark are considered committed.
	SetHighWatermark(hw int64)

	// HighWatermark returns the high watermark for the log.
	HighWatermark() int64

	// Append writes the given batch of messages to the log and returns their
	// corresponding offsets in the log.
	Append(msg []*proto.Message) ([]int64, error)

	// AppendMessageSet writes the given message set data to the log and
	// returns the corresponding offsets in the log.
	AppendMessageSet(ms []byte) ([]int64, error)

	// Close closes each log segment file and stops the background goroutine
	// checkpointing the high watermark to disk.
	Close() error
}

CommitLog is the durable write-ahead log interface used to back each stream.

type Config

type Config struct {
	Host                string
	Port                int
	LogLevel            uint32
	NoLog               bool
	DataDir             string
	BatchMaxMessages    int
	BatchWaitTime       time.Duration
	MetadataCacheMaxAge time.Duration
	NATS                nats.Options
	Log                 LogConfig
	Clustering          ClusteringConfig
}

Config contains all settings for a Liftbridge Server.

func NewConfig

func NewConfig(configFile string) (*Config, error)

NewConfig creates a new Config with default settings and applies any settings from the given configuration file.

func NewDefaultConfig

func NewDefaultConfig() *Config

NewDefaultConfig creates a new Config with default settings.

type LogConfig

type LogConfig struct {
	RetentionMaxBytes int64
	SegmentMaxBytes   int64
	Compact           bool
}

LogConfig contains settings for controlling the message log for a stream.

type Server

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

Server is the main Liftbridge object. Create it by calling New or RunServerWithConfig.

func New

func New(config *Config) *Server

New creates a new Server with the given configuration. Call Start to run the Server.

func RunServerWithConfig

func RunServerWithConfig(config *Config) (*Server, error)

RunServerWithConfig creates and starts a new Server with the given configuration. It returns an error if the Server failed to start.

func (*Server) Apply

func (s *Server) Apply(l *raft.Log) interface{}

Apply applies a Raft log entry to the controller FSM. This is invoked by Raft once a log entry is committed. It returns a value which will be made available on the ApplyFuture returned by Raft.Apply if that method was called on the same Raft node as the FSM.

Note that, on restart, this can be called for entries that have already been committed to Raft as part of the recovery process. As such, this should be an idempotent call.

func (*Server) Restore

func (s *Server) Restore(snapshot io.ReadCloser) error

Restore is used to restore an FSM from a snapshot. It is not called concurrently with any other command. The FSM must discard all previous state.

func (*Server) Snapshot

func (s *Server) Snapshot() (raft.FSMSnapshot, error)

Snapshot is used to support log compaction. This call should return an FSMSnapshot which can be used to save a point-in-time snapshot of the FSM. Apply and Snapshot are not called in multiple threads, but Apply will be called concurrently with Persist. This means the FSM should be implemented in a fashion that allows for concurrent updates while a snapshot is happening.

func (*Server) Start

func (s *Server) Start() error

Start the Server. This is not a blocking call. It will return an error if the Server cannot start properly.

func (*Server) Stop

func (s *Server) Stop() error

Stop will attempt to gracefully shut the Server down by signaling the stop and waiting for all goroutines to return.