raftlease

Documentation

Index

• Constants
• func RecoverError(resp *ResponseError) error
• type Command
  • func (c *Command) LeaseKey() lease.Key
  • func (c *Command) Marshal() ([]byte, error)
  • func (c *Command) String() string
  • 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) LeaseGroup(getLocalTime func() time.Time, namespace, modelUUID string) map[lease.Key]lease.Info
  • 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, stop <-chan struct{}) error
  • func (s *Store) ClaimLease(key lease.Key, req lease.Request, stop <-chan struct{}) error
  • func (s *Store) Collect(ch chan<- prometheus.Metric)
  • func (s *Store) Describe(ch chan<- *prometheus.Desc)
  • func (s *Store) ExtendLease(key lease.Key, req lease.Request, stop <-chan struct{}) error
  • func (s *Store) LeaseGroup(namespace, modelUUID string) map[lease.Key]lease.Info
  • func (s *Store) Leases(keys ...lease.Key) map[lease.Key]lease.Info
  • func (s *Store) PinLease(key lease.Key, entity string, stop <-chan struct{}) error
  • func (s *Store) Pinned() map[lease.Key][]string
  • func (s *Store) RevokeLease(key lease.Key, holder string, stop <-chan struct{}) error
  • func (s *Store) UnpinLease(key lease.Key, entity string, stop <-chan struct{}) 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"

	// OperationRevoke denotes revoking an existing lease.
	OperationRevoke = "revoke"

	// 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 (*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) String

func (c *Command) String() string

String implements fmt.Stringer for the Command type.

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

func (f *FSM) LeaseGroup(getLocalTime func() time.Time, namespace, modelUUID string) map[lease.Key]lease.Info

LeaseGroup returns all leases matching the namespace and model - when there are many models this is more efficient than getting all the leases and filtering by model.

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 and LeaseGroup receive 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
	LeaseGroup(func() time.Time, string, string) 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, stop <-chan struct{}) error

Advance is part of globalclock.Updater.

func (*Store) ClaimLease

func (s *Store) ClaimLease(key lease.Key, req lease.Request, stop <-chan struct{}) 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) ExtendLease

func (s *Store) ExtendLease(key lease.Key, req lease.Request, stop <-chan struct{}) error

ExtendLease is part of lease.Store.

func (*Store) LeaseGroup

func (s *Store) LeaseGroup(namespace, modelUUID string) map[lease.Key]lease.Info

LeaseGroup 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, stop <-chan struct{}) 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) RevokeLease

func (s *Store) RevokeLease(key lease.Key, holder string, stop <-chan struct{}) error

RevokeLease is part of lease.Store.

func (*Store) UnpinLease

func (s *Store) UnpinLease(key lease.Key, entity string, stop <-chan struct{}) 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.