cmdlog

Documentation

Index

• Variables
• func Replay(c nextCursor, jc job.ControllerInterface) error
• func StartCleaningCycle(cleaner cleaner, fn captain.CleanFn, interval uint) error
• type CircuitBreaker
  • func (c *CircuitBreaker) Close()
  • func (c *CircuitBreaker) Open()
  • func (c *CircuitBreaker) State() uint32
• type CircuitBreakerAppender
  • func NewCircuitBreakerAppender(breaker *CircuitBreaker, app appender) *CircuitBreakerAppender
  • func (c *CircuitBreakerAppender) Append(b []byte) error
• type CircuitBreakerHandler
  • func (h *CircuitBreakerHandler) Exec(cmd *prot.Cmd) ([]byte, error)
• type CircuitBreakerRouter
  • func NewCircuitBreakerRouter(breaker *CircuitBreaker, r server.Router) *CircuitBreakerRouter
  • func (c *CircuitBreakerRouter) Handler(cmd string) server.Handler
• type CommandCleaner
  • func NewWarmedCommandCleaner(reg *job.Registry, cur segmentCursor) (*CommandCleaner, error)
  • func (c *CommandCleaner) Clean(path string, r *captain.Record) (bool, error)
• type ControllerProxy
  • func NewControllerProxy(a appender, c job.ControllerInterface) *ControllerProxy
  • func (c *ControllerProxy) Add(j *job.Job) error
  • func (c *ControllerProxy) Complete(id job.ID, result []byte) error
  • func (c *ControllerProxy) Delete(id job.ID) error
  • func (c *ControllerProxy) Expire(id job.ID)
  • func (c *ControllerProxy) ExpireFunc() func(job.ID)
  • func (c *ControllerProxy) Fail(id job.ID, result []byte) error
  • func (c *ControllerProxy) HandleExpire(f func(job.ID))
  • func (c *ControllerProxy) HandleTimeoutAttempt(f func(job.ID))
  • func (c *ControllerProxy) Lease(names []string, timeout uint32) (*job.Job, error)
  • func (c *ControllerProxy) Run(j *job.Job, timeout uint32) (*job.RunResult, error)
  • func (c *ControllerProxy) Schedule(j *job.Job) error
  • func (c *ControllerProxy) StartAttempt(id job.ID) error
  • func (c *ControllerProxy) TimeoutAttempt(id job.ID)
  • func (c *ControllerProxy) TimeoutAttemptFunc() func(job.ID)

Variables

var MagicHeader = &captain.MagicHeader{Magic: 0x77717771, Version: 1}

MagicHeader represents the header all workq log files should use.

Functions

func Replay

func Replay(c nextCursor, jc job.ControllerInterface) error

Replay iterates through all records and replays them with the job controller. Stops on the first error. Complete, Fail, Delete commands have an exception for NOT-FOUND errors. Procesing proceeds for NOT-FOUND as the jobs they are targeted for may have already expired and will be cleaned by the Cleaner.

func StartCleaningCycle

func StartCleaningCycle(cleaner cleaner, fn captain.CleanFn, interval uint) error

StartCleaningCycle initiates the cleaning process in the background with the specified interval. A warmup is performed to retrieve the average job expiration times of each segment file. This prevents a large influx of rewrites on existing segment files if a "cold" cleaning was performed since we would not know how many records required cleaning.

Types

type CircuitBreaker

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

CircuitBreaker provides an atomic Open/Close breaker.

func (*CircuitBreaker) Close

func (c *CircuitBreaker) Close()

Close sets the breaker closed aotmically.

func (*CircuitBreaker) Open

func (c *CircuitBreaker) Open()

Open sets the breaker open atomically.

func (*CircuitBreaker) State

func (c *CircuitBreaker) State() uint32

State returns the current state atomically.

type CircuitBreakerAppender

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

CircuitBreakerAppender wraps and proxies an Appender with a CircuitBreaker. If an append fails, the CircuitBreaker is open, and the failed append will be retried in a recovery loop forever. Any further Appends will be block until the failed one has succeeded.

func NewCircuitBreakerAppender

func NewCircuitBreakerAppender(breaker *CircuitBreaker, app appender) *CircuitBreakerAppender

func (*CircuitBreakerAppender) Append

func (c *CircuitBreakerAppender) Append(b []byte) error

type CircuitBreakerHandler

type CircuitBreakerHandler struct{}

CircuitBreakerHandler implements a Handler that responds with a SERVER-ERROR that states the server is in read-only mode.

func (*CircuitBreakerHandler) Exec

func (h *CircuitBreakerHandler) Exec(cmd *prot.Cmd) ([]byte, error)

type CircuitBreakerRouter

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

CircuitBreakerRouter proxies an existing Server.Router with respect for the breaker's state. When open, all commands respond with a SERVER-ERROR notifying clients that the server is in read-only mode. When closed, all commands are processed normally through the proxied router.

func NewCircuitBreakerRouter

func NewCircuitBreakerRouter(breaker *CircuitBreaker, r server.Router) *CircuitBreakerRouter

func (*CircuitBreakerRouter) Handler

func (c *CircuitBreakerRouter) Handler(cmd string) server.Handler

Handler by command name

type CommandCleaner

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

CommandCleaner implements the captain.CleanFn interface under its Clean method. Cleans only when the average expiration of the records has been met, approximately halving a segment file on every clean.

func NewWarmedCommandCleaner

func NewWarmedCommandCleaner(reg *job.Registry, cur segmentCursor) (*CommandCleaner, error)

func (*CommandCleaner) Clean

func (c *CommandCleaner) Clean(path string, r *captain.Record) (bool, error)

Clean returns a bool if the related job within the record should be deleted. A job is targeted for deletion once it has passed its expiration.

type ControllerProxy

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

ControllerProxy intercepts job.Controller methods that change state and records a cmlog entry before returning to the caller.

func NewControllerProxy

func NewControllerProxy(a appender, c job.ControllerInterface) *ControllerProxy

NewControllerProxy returns a new ControllerProxy and wraps the specified job controller.

func (*ControllerProxy) Add

func (c *ControllerProxy) Add(j *job.Job) error

func (*ControllerProxy) Complete

func (c *ControllerProxy) Complete(id job.ID, result []byte) error

func (*ControllerProxy) Delete

func (c *ControllerProxy) Delete(id job.ID) error

func (*ControllerProxy) Expire

func (c *ControllerProxy) Expire(id job.ID)

func (*ControllerProxy) ExpireFunc

func (c *ControllerProxy) ExpireFunc() func(job.ID)

func (*ControllerProxy) Fail

func (c *ControllerProxy) Fail(id job.ID, result []byte) error

func (*ControllerProxy) HandleExpire

func (c *ControllerProxy) HandleExpire(f func(job.ID))

func (*ControllerProxy) HandleTimeoutAttempt

func (c *ControllerProxy) HandleTimeoutAttempt(f func(job.ID))

func (*ControllerProxy) Lease

func (c *ControllerProxy) Lease(names []string, timeout uint32) (*job.Job, error)

func (*ControllerProxy) Run

func (c *ControllerProxy) Run(j *job.Job, timeout uint32) (*job.RunResult, error)

func (*ControllerProxy) Schedule

func (c *ControllerProxy) Schedule(j *job.Job) error

func (*ControllerProxy) StartAttempt

func (c *ControllerProxy) StartAttempt(id job.ID) error

func (*ControllerProxy) TimeoutAttempt

func (c *ControllerProxy) TimeoutAttempt(id job.ID)

func (*ControllerProxy) TimeoutAttemptFunc

func (c *ControllerProxy) TimeoutAttemptFunc() func(job.ID)