cluster

Documentation

Index

• type Cluster
  • func New(nodeStore NodeGroupStore, nodeNotifyQueue NodeMessageQueue, ...) *Cluster
  • func (cluster *Cluster) Close()
• type Labels
• type MemoGroupStore
  • func (store *MemoGroupStore) ReadNodeGroup(ctx context.Context) (*NodeGroup, error)
  • func (store *MemoGroupStore) SaveNodes(ctx context.Context, group *NodeGroup) error
• type MemoMessageQueue
  • func NewMemoMessageQueue() *MemoMessageQueue
  • func (queue *MemoMessageQueue) Close() error
  • func (queue *MemoMessageQueue) NewScheduleQueue(nodeID string) *MemoScheduleQueue
  • func (queue *MemoMessageQueue) PushNodeTransaction(ctx context.Context, transaction Transaction) error
  • func (queue *MemoMessageQueue) WaitMessages(ctx context.Context) ([]NodeMessage, error)
• type MemoScheduleQueue
  • func (queue *MemoScheduleQueue) Ack(ctx context.Context, node Node) error
  • func (queue *MemoScheduleQueue) Close() error
  • func (queue *MemoScheduleQueue) WaitEvent(ctx context.Context) (*NodeScheduleEvent, error)
• type Node
• type NodeAgent
  • func NewNodeAgent(store Node, eventStream ScheduleEventQueue) *NodeAgent
  • func (manager *NodeAgent) AddUnits(ctx context.Context, units ...Unit) error
  • func (manager *NodeAgent) GetCurrentNode(ctx context.Context) (Node, error)
  • func (manager *NodeAgent) GetCurrentSchedule(ctx context.Context) (expect *UnitSet, err error)
  • func (manager *NodeAgent) RemoveUnits(ctx context.Context, keys []string) error
  • func (manager *NodeAgent) SendCurrentInfo(ctx context.Context) error
  • func (manager *NodeAgent) SetLeaveCluster(ctx context.Context) error
  • func (manager *NodeAgent) SetLiveInfo(ctx context.Context, available float64) error
  • func (manager *NodeAgent) WaitSchedule(ctx context.Context) (expect *UnitSet, err error)
• type NodeEvent
• type NodeEventQueue
  • func (q *NodeEventQueue) Peek(ctx context.Context) (*NodeEvent, error)
  • func (q *NodeEventQueue) Reset()
• type NodeGroup
  • func NewNodeGroup(nodes []Node) *NodeGroup
  • func (group *NodeGroup) GetNodeByID(id string) (Node, bool)
  • func (group *NodeGroup) GetNodes() []Node
  • func (group *NodeGroup) Size() int
  • func (group *NodeGroup) UpdateNode(node Node)
• type NodeGroupStore
• type NodeMessage
• type NodeMessageQueue
• type NodeObserver
  • func (observer *NodeObserver) GetEventQueue(ctx context.Context) *NodeEventQueue
  • func (observer *NodeObserver) GetNodeGroup(ctx context.Context) (*NodeGroup, error)
• type NodeSchedule
• type NodeScheduleEvent
• type NotificationAcceptor
• type ScheduleEventQueue
• type ScheduleTrigger
• type Scheduler
• type SchedulerFunc
  • func (f SchedulerFunc) WaitSchedulerTrigger(ctx context.Context) ([]NodeSchedule, error)
• type Transaction
• type Unit
• type UnitSet
  • func NewUnitSet() *UnitSet
  • func (set *UnitSet) Copy() *UnitSet
  • func (set *UnitSet) Count() int
  • func (set *UnitSet) Get(key string) (Unit, bool)
  • func (set *UnitSet) List() []Unit
  • func (set *UnitSet) Remove(key string)
  • func (set *UnitSet) Update(unit Unit)

Types

type Cluster

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

func New

func New(
	nodeStore NodeGroupStore,
	nodeNotifyQueue NodeMessageQueue,
	schedulers Scheduler,
	records record.Factory) *Cluster

func (*Cluster) Close

func (cluster *Cluster) Close()

type Labels

type Labels textproto.MIMEHeader

type MemoGroupStore

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

func (*MemoGroupStore) ReadNodeGroup

func (store *MemoGroupStore) ReadNodeGroup(ctx context.Context) (*NodeGroup, error)

func (*MemoGroupStore) SaveNodes

func (store *MemoGroupStore) SaveNodes(ctx context.Context, group *NodeGroup) error

type MemoMessageQueue

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

func NewMemoMessageQueue

func NewMemoMessageQueue() *MemoMessageQueue

func (*MemoMessageQueue) Close

func (queue *MemoMessageQueue) Close() error

func (*MemoMessageQueue) NewScheduleQueue

func (queue *MemoMessageQueue) NewScheduleQueue(nodeID string) *MemoScheduleQueue

func (*MemoMessageQueue) PushNodeTransaction

func (queue *MemoMessageQueue) PushNodeTransaction(ctx context.Context, transaction Transaction) error

func (*MemoMessageQueue) WaitMessages

func (queue *MemoMessageQueue) WaitMessages(ctx context.Context) ([]NodeMessage, error)

type MemoScheduleQueue

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

func (*MemoScheduleQueue) Ack

func (queue *MemoScheduleQueue) Ack(ctx context.Context, node Node) error

func (*MemoScheduleQueue) Close

func (queue *MemoScheduleQueue) Close() error

func (*MemoScheduleQueue) WaitEvent

func (queue *MemoScheduleQueue) WaitEvent(ctx context.Context) (*NodeScheduleEvent, error)

type Node

type Node struct {
	ID            string    `bson:"id"`
	Seq           int64     `bson:"seq"`
	LastKeepLive  time.Time `bson:"keepLive"`
	LastAvailable float64   `bson:"available"`
	Units         UnitSet   `bson:"units"`
	Labels        Labels    `bson:"labels"`
}

type NodeAgent

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

func NewNodeAgent

func NewNodeAgent(store Node, eventStream ScheduleEventQueue) *NodeAgent

func (*NodeAgent) AddUnits

func (manager *NodeAgent) AddUnits(ctx context.Context, units ...Unit) error

func (*NodeAgent) GetCurrentNode

func (manager *NodeAgent) GetCurrentNode(ctx context.Context) (Node, error)

func (*NodeAgent) GetCurrentSchedule

func (manager *NodeAgent) GetCurrentSchedule(ctx context.Context) (expect *UnitSet, err error)

func (*NodeAgent) RemoveUnits

func (manager *NodeAgent) RemoveUnits(ctx context.Context, keys []string) error

func (*NodeAgent) SendCurrentInfo

func (manager *NodeAgent) SendCurrentInfo(ctx context.Context) error

func (*NodeAgent) SetLeaveCluster

func (manager *NodeAgent) SetLeaveCluster(ctx context.Context) error

func (*NodeAgent) SetLiveInfo

func (manager *NodeAgent) SetLiveInfo(ctx context.Context, available float64) error

func (*NodeAgent) WaitSchedule

func (manager *NodeAgent) WaitSchedule(ctx context.Context) (expect *UnitSet, err error)

type NodeEvent

type NodeEvent struct {
	Seq    int64
	Node   Node
	Origin *Node
}

type NodeEventQueue

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

func (*NodeEventQueue) Peek

func (q *NodeEventQueue) Peek(ctx context.Context) (*NodeEvent, error)

func (*NodeEventQueue) Reset

func (q *NodeEventQueue) Reset()

type NodeGroup

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

func NewNodeGroup

func NewNodeGroup(nodes []Node) *NodeGroup

func (*NodeGroup) GetNodeByID

func (group *NodeGroup) GetNodeByID(id string) (Node, bool)

func (*NodeGroup) GetNodes

func (group *NodeGroup) GetNodes() []Node

func (*NodeGroup) Size

func (group *NodeGroup) Size() int

func (*NodeGroup) UpdateNode

func (group *NodeGroup) UpdateNode(node Node)

type NodeGroupStore

type NodeGroupStore interface {
	ReadNodeGroup(ctx context.Context) (*NodeGroup, error)
	SaveNodes(ctx context.Context, group *NodeGroup) error
}

type NodeMessage

type NodeMessage struct {
	Timestamp time.Time
	Node      Node
}

type NodeMessageQueue

type NodeMessageQueue interface {
	WaitMessages(ctx context.Context) ([]NodeMessage, error)
	PushNodeTransaction(ctx context.Context, transaction Transaction) error
	Close() error
}

type NodeObserver

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

func (*NodeObserver) GetEventQueue

func (observer *NodeObserver) GetEventQueue(ctx context.Context) *NodeEventQueue

func (*NodeObserver) GetNodeGroup

func (observer *NodeObserver) GetNodeGroup(ctx context.Context) (*NodeGroup, error)

type NodeSchedule

type NodeSchedule struct {
	NodeID     string
	UpdateUnit []Unit
	RemoveUnit []string
}

type NodeScheduleEvent

type NodeScheduleEvent struct {
	Schedule NodeSchedule
	Seq      int64
}

type NotificationAcceptor

type NotificationAcceptor interface {
	Accept(ctx context.Context, notification NodeMessage) error
	WaitScheduleTask(ctx context.Context) ([]NodeSchedule, error)
}

type ScheduleEventQueue

type ScheduleEventQueue interface {
	WaitEvent(ctx context.Context) (*NodeScheduleEvent, error)
	Ack(ctx context.Context, node Node) error
	Close() error
}

type ScheduleTrigger

type ScheduleTrigger interface {
	WaitSchedulerTrigger(ctx context.Context) ([]NodeSchedule, error)
}

type Scheduler

type Scheduler interface {
	StartSchedule(ctx context.Context, observer *NodeObserver) (ScheduleTrigger, error)
}

type SchedulerFunc

type SchedulerFunc func(ctx context.Context) ([]NodeSchedule, error)

func (SchedulerFunc) WaitSchedulerTrigger

func (f SchedulerFunc) WaitSchedulerTrigger(ctx context.Context) ([]NodeSchedule, error)

type Transaction

type Transaction struct {
	Seq       int64
	Schedules []NodeSchedule
}

type Unit

type Unit struct {
	Key    string
	Labels Labels
	Status int32 // for customer change
}

type UnitSet

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

func NewUnitSet

func NewUnitSet() *UnitSet

func (*UnitSet) Copy

func (set *UnitSet) Copy() *UnitSet

func (*UnitSet) Count

func (set *UnitSet) Count() int

func (*UnitSet) Get

func (set *UnitSet) Get(key string) (Unit, bool)

func (*UnitSet) List

func (set *UnitSet) List() []Unit

func (*UnitSet) Remove

func (set *UnitSet) Remove(key string)

func (*UnitSet) Update

func (set *UnitSet) Update(unit Unit)