raftlease

Documentation

Index

• Constants
• func RecoverError(resp *ResponseError) error
• type Command
  • func UnmarshalCommand(data []byte) (*Command, error)
  • func (c *Command) LeaseKey() lease.Key
  • func (c *Command) Marshal() ([]byte, error)
  • func (c *Command) Validate() error
• type FSM
  • func NewFSM() *FSM
  • func (f *FSM) Apply(log *raft.Log) interface{}
  • func (f *FSM) GlobalTime() time.Time
  • func (f *FSM) Leases(getLocalTime func() time.Time, keys ...lease.Key) map[lease.Key]lease.Info
  • func (f *FSM) Pinned() map[lease.Key][]string
  • func (f *FSM) Restore(reader io.ReadCloser) error
  • func (f *FSM) Snapshot() (raft.FSMSnapshot, error)
• type FSMResponse
• type ForwardRequest
• type ForwardResponse
• type NotifyTarget
• type ReadonlyFSM
• type ResponseError
  • func AsResponseError(err error) *ResponseError
• type Snapshot
  • func (s *Snapshot) Persist(sink raft.SnapshotSink) (err error)
  • func (s *Snapshot) Release()
• type SnapshotEntry
• type SnapshotKey
• type Store
  • func NewStore(config StoreConfig) *Store
  • func (s *Store) Advance(duration time.Duration) error
  • func (*Store) Autoexpire() bool
  • func (s *Store) ClaimLease(key lease.Key, req lease.Request) error
  • func (s *Store) Collect(ch chan<- prometheus.Metric)
  • func (s *Store) Describe(ch chan<- *prometheus.Desc)
  • func (s *Store) ExpireLease(key lease.Key) error
  • func (s *Store) ExtendLease(key lease.Key, req lease.Request) error
  • func (s *Store) Leases(keys ...lease.Key) map[lease.Key]lease.Info
  • func (s *Store) PinLease(key lease.Key, entity string) error
  • func (s *Store) Pinned() map[lease.Key][]string
  • func (s *Store) Refresh() error
  • func (s *Store) UnpinLease(key lease.Key, entity string) error
• type StoreConfig
• type TrapdoorFunc

Constants

const (
	// CommandVersion is the current version of the command format. If
	// this changes then we need to be sure that reading and applying
	// commands for previous versions still works.
	CommandVersion = 1

	// SnapshotVersion is the current version of the snapshot
	// format. Similarly, changes to the snapshot representation need
	// to be backward-compatible.
	SnapshotVersion = 1

	// OperationClaim denotes claiming a new lease.
	OperationClaim = "claim"

	// OperationExtend denotes extending an already-held lease.
	OperationExtend = "extend"

	// OperationSetTime denotes updating stored global time (which
	// will also remove any expired leases).
	OperationSetTime = "setTime"

	// OperationPin pins a lease, preventing it from expiring
	// until it is unpinned.
	OperationPin = "pin"

	// OperationUnpin unpins a lease, restoring normal
	// lease expiry behaviour.
	OperationUnpin = "unpin"
)

Functions

func RecoverError

func RecoverError(resp *ResponseError) error

RecoverError converts a ResponseError back into the specific error it represents, or into a generic error if it wasn't one of the singleton errors handled.

Types

type Command

type Command struct {
	// Version of the command format, in case it changes and we need
	// to handle multiple formats.
	Version int `yaml:"version"`

	// Operation is one of claim, extend, expire or setTime.
	Operation string `yaml:"operation"`

	// Namespace is the kind of lease.
	Namespace string `yaml:"namespace,omitempty"`

	// ModelUUID identifies the model the lease belongs to.
	ModelUUID string `yaml:"model-uuid,omitempty"`

	// Lease is the name of the lease the command affects.
	Lease string `yaml:"lease,omitempty"`

	// Holder is the name of the party claiming or extending the
	// lease.
	Holder string `yaml:"holder,omitempty"`

	// Duration is how long the lease should last.
	Duration time.Duration `yaml:"duration,omitempty"`

	// OldTime is the previous time for time updates (to avoid
	// applying stale ones).
	OldTime time.Time `yaml:"old-time,omitempty"`

	// NewTime is the time to store as the global time.
	NewTime time.Time `yaml:"new-time,omitempty"`

	// PinEntity is a tag representing an entity concerned
	// with a pin or unpin operation.
	PinEntity string `yaml:"pin-entity,omitempty"`
}

Command captures the details of an operation to be run on the FSM.

func UnmarshalCommand

func UnmarshalCommand(data []byte) (*Command, error)

UnmarshalCommand converts a marshalled command []byte into a command.

func (*Command) LeaseKey

func (c *Command) LeaseKey() lease.Key

LeaseKey makes a lease key from the fields in the command.

func (*Command) Marshal

func (c *Command) Marshal() ([]byte, error)

Marshal converts this command to a byte slice.

func (*Command) Validate

func (c *Command) Validate() error

Validate checks that the command describes a valid state change.

type FSM

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

FSM stores the state of leases in the system.

func NewFSM

func NewFSM() *FSM

NewFSM returns a new FSM to store lease information.

func (*FSM) Apply

func (f *FSM) Apply(log *raft.Log) interface{}

Apply is part of raft.FSM.

func (*FSM) GlobalTime

func (f *FSM) GlobalTime() time.Time

GlobalTime returns the FSM's internal time.

func (*FSM) Leases

func (f *FSM) Leases(getLocalTime func() time.Time, keys ...lease.Key) map[lease.Key]lease.Info

Leases gets information about all of the leases in the system, optionally filtered by the input lease keys.

func (*FSM) Pinned

func (f *FSM) Pinned() map[lease.Key][]string

Pinned returns all of the currently known lease pins and applications requiring the pinned behaviour.

func (*FSM) Restore

func (f *FSM) Restore(reader io.ReadCloser) error

Restore is part of raft.FSM.

func (*FSM) Snapshot

func (f *FSM) Snapshot() (raft.FSMSnapshot, error)

Snapshot is part of raft.FSM.

type FSMResponse

type FSMResponse interface {
	// Error is a lease error (rather than anything to do with the
	// raft machinery).
	Error() error

	// Notify tells the target what changes occurred because of the
	// applied command.
	Notify(NotifyTarget)
}

FSMResponse defines what will be available on the return value from FSM apply calls.

type ForwardRequest

type ForwardRequest struct {
	Command       string `yaml:"command"`
	ResponseTopic string `yaml:"response-topic"`
}

ForwardRequest is a message sent over the hub to the raft forwarder (only running on the raft leader node).

type ForwardResponse

type ForwardResponse struct {
	Error *ResponseError `yaml:"error"`
}

ForwardResponse is the response sent back from the raft forwarder.

type NotifyTarget

type NotifyTarget interface {
	// Claimed will be called when a new lease has been claimed. Not
	// allowed to return an error because this is purely advisory -
	// the lease claim has still occurred, whether or not the callback
	// succeeds.
	Claimed(lease.Key, string)

	// Expired will be called when an existing lease has expired. Not
	// allowed to return an error because this is purely advisory.
	Expired(lease.Key)
}

NotifyTarget defines methods needed to keep an external database updated with who holds leases. (In non-test code the notify target will generally be the state DB.)

type ReadonlyFSM

type ReadonlyFSM interface {
	// Leases receives a func for retrieving time, because it needs to be
	// determined after potential lock-waiting to be accurate.
	Leases(func() time.Time, ...lease.Key) map[lease.Key]lease.Info
	GlobalTime() time.Time
	Pinned() map[lease.Key][]string
}

ReadonlyFSM defines the methods of the lease FSM the store can use - any writes must go through the hub.

type ResponseError

type ResponseError struct {
	Message string `yaml:"message"`
	Code    string `yaml:"code"`
}

ResponseError is used for sending error values back to the lease store via the hub.

func AsResponseError

func AsResponseError(err error) *ResponseError

AsResponseError returns a *ResponseError that can be sent back over the hub in response to a forwarded FSM command.

type Snapshot

type Snapshot struct {
	Version    int                           `yaml:"version"`
	Entries    map[SnapshotKey]SnapshotEntry `yaml:"entries"`
	Pinned     map[SnapshotKey][]string      `yaml:"pinned"`
	GlobalTime time.Time                     `yaml:"global-time"`
}

Snapshot defines the format of the FSM snapshot.

func (*Snapshot) Persist

func (s *Snapshot) Persist(sink raft.SnapshotSink) (err error)

Persist is part of raft.FSMSnapshot.

func (*Snapshot) Release

func (s *Snapshot) Release()

Release is part of raft.FSMSnapshot.

type SnapshotEntry

type SnapshotEntry struct {
	Holder   string        `yaml:"holder"`
	Start    time.Time     `yaml:"start"`
	Duration time.Duration `yaml:"duration"`
}

SnapshotEntry defines the format of a lease entry in a snapshot.

type SnapshotKey

type SnapshotKey struct {
	Namespace string `yaml:"namespace"`
	ModelUUID string `yaml:"model-uuid"`
	Lease     string `yaml:"lease"`
}

SnapshotKey defines the format of a lease key in a snapshot.

type Store

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

Store manages a raft FSM and forwards writes through a pubsub hub.

func NewStore

func NewStore(config StoreConfig) *Store

NewStore returns a core/lease.Store that manages leases in Raft.

func (*Store) Advance

func (s *Store) Advance(duration time.Duration) error

Advance is part of globalclock.Updater.

func (*Store) Autoexpire

func (*Store) Autoexpire() bool

Autoexpire is part of lease.Store.

func (*Store) ClaimLease

func (s *Store) ClaimLease(key lease.Key, req lease.Request) error

ClaimLease is part of lease.Store.

func (*Store) Collect

func (s *Store) Collect(ch chan<- prometheus.Metric)

Collect is part of prometheus.Collector.

func (*Store) Describe

func (s *Store) Describe(ch chan<- *prometheus.Desc)

Describe is part of prometheus.Collector.

func (*Store) ExpireLease

func (s *Store) ExpireLease(key lease.Key) error

ExpireLease is part of lease.Store.

func (*Store) ExtendLease

func (s *Store) ExtendLease(key lease.Key, req lease.Request) error

ExtendLease is part of lease.Store.

func (*Store) Leases

func (s *Store) Leases(keys ...lease.Key) map[lease.Key]lease.Info

Leases is part of lease.Store.

func (*Store) PinLease

func (s *Store) PinLease(key lease.Key, entity string) error

PinLease is part of lease.Store.

func (*Store) Pinned

func (s *Store) Pinned() map[lease.Key][]string

Pinned is part of the Store interface.

func (*Store) Refresh

func (s *Store) Refresh() error

Refresh is part of lease.Store.

func (*Store) UnpinLease

func (s *Store) UnpinLease(key lease.Key, entity string) error

UnpinLease is part of lease.Store.

type StoreConfig

type StoreConfig struct {
	FSM           ReadonlyFSM
	Hub           *pubsub.StructuredHub
	Trapdoor      TrapdoorFunc
	RequestTopic  string
	ResponseTopic func(requestID uint64) string

	Clock          clock.Clock
	ForwardTimeout time.Duration
}

StoreConfig holds resources and settings needed to run the Store.

type TrapdoorFunc

type TrapdoorFunc func(lease.Key, string) lease.Trapdoor

TrapdoorFunc returns a trapdoor to be attached to lease details for use by clients. This is intended to hold assertions that can be added to state transactions to ensure the lease is still held when the transaction is applied.