server

Documentation

Index

• Constants
• Variables
• func GetLogLevel(level string) (uint32, error)
• type ActivityStreamConfig
• type ClusteringConfig
• type Config
  • func NewConfig(configFile string) (*Config, error)
  • func NewDefaultConfig() *Config
  • func (c Config) GetConnectionAddress() HostPort
  • func (c Config) GetListenAddress() HostPort
• type HostPort
• type Server
  • func New(config *Config) *Server
  • func RunServerWithConfig(config *Config) (*Server, error)
  • func (s *Server) Apply(l *raft.Log) interface{}
  • func (s *Server) IsLeader() bool
  • func (s *Server) IsRunning() bool
  • func (s *Server) Restore(snapshot io.ReadCloser) error
  • func (s *Server) Snapshot() (raft.FSMSnapshot, error)
  • func (s *Server) Start() (err error)
  • func (s *Server) Stop() error
• type StreamsConfig
  • func (l *StreamsConfig) ApplyOverrides(c *proto.StreamConfig)
  • func (l StreamsConfig) RetentionString() string

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
)

const Version = "v1.2.0"

Version of the Liftbridge server.

Variables

var (
	// ErrStreamExists is returned by CreateStream when attempting to create a
	// stream that already exists.
	ErrStreamExists = errors.New("stream already exists")

	// ErrStreamNotFound is returned by DeleteStream/PauseStream when
	// attempting to delete/pause a stream that does not exist.
	ErrStreamNotFound = errors.New("stream does not exist")

	// ErrPartitionNotFound is returned by PauseStream when attempting to pause
	// a stream partition that does not exist.
	ErrPartitionNotFound = errors.New("partition does not exist")
)

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 ActivityStreamConfig

type ActivityStreamConfig struct {
	Enabled          bool
	PublishTimeout   time.Duration
	PublishAckPolicy client.AckPolicy
}

ActivityStreamConfig contains settings for controlling activity stream behavior.

type ClusteringConfig

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

ClusteringConfig contains settings for controlling cluster behavior.

type Config

type Config struct {
	Listen              HostPort
	Host                string
	Port                int
	LogLevel            uint32
	LogRecovery         bool
	LogRaft             bool
	LogSilent           bool
	DataDir             string
	BatchMaxMessages    int
	BatchMaxTime        time.Duration
	MetadataCacheMaxAge time.Duration
	TLSKey              string
	TLSCert             string
	TLSClientAuth       bool
	TLSClientAuthCA     string
	NATS                nats.Options
	Streams             StreamsConfig
	Clustering          ClusteringConfig
	ActivityStream      ActivityStreamConfig
}

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.

func (Config) GetConnectionAddress

func (c Config) GetConnectionAddress() HostPort

GetConnectionAddress returns the host if specified and listen otherwise.

func (Config) GetListenAddress

func (c Config) GetListenAddress() HostPort

GetListenAddress returns the address and port to listen to.

type HostPort

type HostPort struct {
	Host string
	Port int
}

HostPort is simple struct to hold parsed listen/addr strings.

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) IsLeader

func (s *Server) IsLeader() bool

IsLeader indicates if the server is currently the metadata leader or not. If consistency is required for an operation, it should be threaded through the Raft cluster since that is the single source of truth. If a server thinks it's leader when it's not, the operation it proposes to the Raft cluster will fail.

func (*Server) IsRunning

func (s *Server) IsRunning() bool

IsRunning indicates if the server is currently running or has been stopped.

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() (err 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.

type StreamsConfig

type StreamsConfig struct {
	RetentionMaxBytes    int64
	RetentionMaxMessages int64
	RetentionMaxAge      time.Duration
	CleanerInterval      time.Duration
	SegmentMaxBytes      int64
	SegmentMaxAge        time.Duration
	Compact              bool
	CompactMaxGoroutines int
}

StreamsConfig contains settings for controlling the message log for streams.

func (*StreamsConfig) ApplyOverrides

func (l *StreamsConfig) ApplyOverrides(c *proto.StreamConfig)

ApplyOverrides applies the values from the StreamConfig protobuf to the StreamsConfig struct. If the value is present in the request's config section, it will be set in StreamsConfig.

func (StreamsConfig) RetentionString

func (l StreamsConfig) RetentionString() string

RetentionString returns a human-readable string representation of the retention policy.