leases

Documentation

Index

• type Config
  • func DefaultOptions() *Config
• type Connector
• type Heartbeat
  • func NewHeartbeat(cfg *Config, schema Schema) *Heartbeat
  • func (h *Heartbeat) Heartbeat(ctx context.Context, db dbtx.DBTX) ([]uint32, error)
• type Heartbeater
• type Leases
  • func New(connector Connector, schema Schema, opts ...Option) (*Leases, error)
  • func NewLeases(cfg *Config, heartbeat Heartbeater, connector Connector) *Leases
  • func (s *Leases) Start(ctx context.Context) error
  • func (s *Leases) Values() []uint32
• type Option
  • func WithHeartbeatInterval(interval time.Duration) Option
  • func WithHeartbeatTimeout(timeout time.Duration) Option
  • func WithLeaseTTL(ttl time.Duration) Option
  • func WithNodeName(nodeName string) Option
  • func WithRand(r *rand.Rand) Option
  • func WithRange(r Range) Option
  • func WithVNodeCount(count uint32) Option
  • func WithValueListener(listener ValueListener) Option
• type Range
  • func (r Range) Len() uint32
  • func (r Range) String() string
  • func (r Range) VNode(ran *rand.Rand) uint32
  • func (r Range) Valid() bool
  • func (r Range) Values() []uint32
• type Report
• type Ring
  • func (ring Ring) Analyze(name string, full Range, count uint32) Report
• type Schema
• type Status
  • func (s Status) String() string
• type VNode
• type ValueListener
• type ValueListenerFunc
  • func (fn ValueListenerFunc) ValuesChanged(values, added, removed []uint32) error

Types

type Config

type Config struct {
	NodeName          string
	Range             Range
	VNodeCount        uint32
	HeartbeatTimeout  time.Duration
	HeartbeatInterval time.Duration
	LeaseTTL          time.Duration
	Rand              *rand.Rand
	// contains filtered or unexported fields
}

func DefaultOptions

func DefaultOptions() *Config

type Connector

type Connector interface {
	AcquireWrite(ctx context.Context) (*pgxpool.Conn, error)
}

type Heartbeat

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

func NewHeartbeat

func NewHeartbeat(cfg *Config, schema Schema) *Heartbeat

func (*Heartbeat) Heartbeat

func (h *Heartbeat) Heartbeat(ctx context.Context, db dbtx.DBTX) ([]uint32, error)

type Heartbeater

type Heartbeater interface {
	Heartbeat(ctx context.Context, conn dbtx.DBTX) ([]uint32, error)
}

type Leases

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

func New

func New(connector Connector, schema Schema, opts ...Option) (*Leases, error)

func NewLeases

func NewLeases(cfg *Config, heartbeat Heartbeater, connector Connector) *Leases

func (*Leases) Start

func (s *Leases) Start(ctx context.Context) error

Start is blocking and keeps the Values up to date.

func (*Leases) Values

func (s *Leases) Values() []uint32

type Option

type Option func(cfg *Config)

func WithHeartbeatInterval

func WithHeartbeatInterval(interval time.Duration) Option

func WithHeartbeatTimeout

func WithHeartbeatTimeout(timeout time.Duration) Option

func WithLeaseTTL

func WithLeaseTTL(ttl time.Duration) Option

func WithNodeName

func WithNodeName(nodeName string) Option

func WithRand

func WithRand(r *rand.Rand) Option

func WithRange

func WithRange(r Range) Option

func WithVNodeCount

func WithVNodeCount(count uint32) Option

func WithValueListener

func WithValueListener(listener ValueListener) Option

type Range

type Range struct {
	From uint32
	To   uint32
}

func (Range) Len

func (r Range) Len() uint32

func (Range) String

func (r Range) String() string

func (Range) VNode

func (r Range) VNode(ran *rand.Rand) uint32

func (Range) Valid

func (r Range) Valid() bool

func (Range) Values

func (r Range) Values() []uint32

type Report

type Report struct {
	Approve      []VNode
	MissingCount uint32
	UsedVNodes   vnodeSet
	Values       []uint32
}

type Ring

type Ring []VNode

func (Ring) Analyze

func (ring Ring) Analyze(name string, full Range, count uint32) Report

type Schema

type Schema interface {
	RefreshLeases(ctx context.Context, db dbtx.DBTX, nodeName string, ttl time.Duration) (Ring, error)
	ApproveLease(ctx context.Context, db dbtx.DBTX, vnodes []uint32) error
	InsertLease(ctx context.Context, db dbtx.DBTX, vnode uint32, name string, ttl time.Duration, status string) error
}

type Status

type Status string

const (
	Pending Status = "PENDING"
	Leased  Status = "LEASED"
)

func (Status) String

func (s Status) String() string

type VNode

type VNode struct {
	VNode    uint32
	NodeName string
	Valid    bool
	Status   Status
}

type ValueListener

type ValueListener interface {
	ValuesChanged(values, added, removed []uint32) error
}

type ValueListenerFunc

type ValueListenerFunc func(values, added, removed []uint32) error

func (ValueListenerFunc) ValuesChanged

func (fn ValueListenerFunc) ValuesChanged(values, added, removed []uint32) error