prophet: github.com/infinivision/prophet Index | Files

package prophet

import "github.com/infinivision/prophet"

Index

Package Files

cfg.go election.go election_option.go errors.go local.go log.go meta.go prophet.go prophet_coordinator.go prophet_etcd.go prophet_handler.go prophet_heartbeat.go prophet_join.go prophet_leader.go prophet_options.go prophet_rpc.go prophet_transport.go runtime.go runtime_container.go runtime_resource.go runtime_update.go scheduler.go scheduler_balancer.go scheduler_balancer_leader.go scheduler_balancer_replica.go scheduler_filter.go scheduler_op_aggregation.go scheduler_op_leader.go scheduler_op_replica.go scheduler_replica.go scheduler_selector.go store.go store_etcd.go util_cache.go util_math.go util_queue.go util_status.go util_task.go util_url.go watcher.go watcher_client.go

Constants

const (
    // DefaultTimeout default timeout
    DefaultTimeout = time.Second * 3
    // DefaultRequestTimeout default request timeout
    DefaultRequestTimeout = 10 * time.Second
    // DefaultSlowRequestTime default slow request time
    DefaultSlowRequestTime = time.Second * 1
)
const (
    // Pending job is wait to running
    Pending = JobState(0)
    // Running job is running
    Running = JobState(1)
    // Cancelling job is cancelling
    Cancelling = JobState(2)
    // Cancelled job is cancelled
    Cancelled = JobState(3)
    // Finished job is complete
    Finished = JobState(4)
    // Failed job is failed when execute
    Failed = JobState(5)
)

Variables

var (
    // ErrSchedulerExisted error with scheduler is existed
    ErrSchedulerExisted = errors.New("scheduler is existed")
    // ErrSchedulerNotFound error with scheduler is not found
    ErrSchedulerNotFound = errors.New("scheduler is not found")
)
var (
    // AddPeer add peer
    AddPeer = ChangePeerType(0)
    // RemovePeer remove peer
    RemovePeer = ChangePeerType(1)
)
var (
    // ErrDisposed is returned when an operation is performed on a disposed
    // queue.
    ErrDisposed = errors.New(`queue: disposed`)

    // ErrTimeout is returned when an applicable queue operation times out.
    ErrTimeout = errors.New(`queue: poll timed out`)

    // ErrEmptyQueue is returned when an non-applicable queue operation was called
    // due to the queue's empty item state
    ErrEmptyQueue = errors.New(`queue: empty queue`)
)
var (
    // EventInit event init
    EventInit = 1 << 1
    // EventResourceCreated event resource created
    EventResourceCreated = 1 << 2
    // EventResourceLeaderChanged event resource leader changed
    EventResourceLeaderChanged = 1 << 3
    // EventResourceChaned event resource changed
    EventResourceChaned = 1 << 4
    // EventContainerCreated  event container create
    EventContainerCreated = 1 << 5
    // EventContainerChanged  event container create
    EventContainerChanged = 1 << 6

    // EventFlagResource all resource event
    EventFlagResource = EventResourceCreated | EventResourceLeaderChanged | EventResourceChaned
    // EventFlagContainer all container event
    EventFlagContainer = EventContainerCreated | EventContainerChanged
    // EventFlagAll all event
    EventFlagAll = 0xffffffff
)
var (
    // ErrJobCancelled error job cancelled
    ErrJobCancelled = errors.New("Job cancelled")
)

func ExecuteInParallel Uses

func ExecuteInParallel(q *Queue, fn func(interface{}))

ExecuteInParallel will (in parallel) call the provided function with each item in the queue until the queue is exhausted. When the queue is exhausted execution is complete and all goroutines will be killed. This means that the queue will be disposed so cannot be used again.

func GeometricMean Uses

func GeometricMean(input Float64Data) (float64, error)

GeometricMean gets the geometric mean for a slice of numbers

func HarmonicMean Uses

func HarmonicMean(input Float64Data) (float64, error)

HarmonicMean gets the harmonic mean for a slice of numbers

func MatchEvent Uses

func MatchEvent(event, flag int) bool

MatchEvent returns the flag has the target event

func Max Uses

func Max(input Float64Data) (max float64, err error)

Max finds the highest number in a slice

func Mean Uses

func Mean(input Float64Data) (float64, error)

Mean gets the average of a slice of numbers

func Median Uses

func Median(input Float64Data) (median float64, err error)

Median gets the median number in a slice of numbers

func Min Uses

func Min(input Float64Data) (min float64, err error)

Min finds the lowest number in a set of data

func Mode Uses

func Mode(input Float64Data) (mode []float64, err error)

Mode gets the mode of a slice of numbers

func PopulationVariance Uses

func PopulationVariance(input Float64Data) (pvar float64, err error)

PopulationVariance finds the amount of variance within a population

func SetLogger Uses

func SetLogger(l Logger)

SetLogger set the log for prophet

func StandardDeviation Uses

func StandardDeviation(input Float64Data) (sdev float64, err error)

StandardDeviation the amount of variation in the dataset

func StandardDeviationPopulation Uses

func StandardDeviationPopulation(input Float64Data) (sdev float64, err error)

StandardDeviationPopulation finds the amount of variation from the population

func Sum Uses

func Sum(input Float64Data) (sum float64, err error)

Sum adds all the numbers of a slice together

type Adapter Uses

type Adapter interface {
    // NewResource return a new resource
    NewResource() Resource
    // NewContainer return a new container
    NewContainer() Container
    // FetchLeaderResources fetch loacle leader resource
    FetchLeaderResources() []uint64
    // FetchResourceHB fetch resource HB
    FetchResourceHB(id uint64) *ResourceHeartbeatReq
    // FetchContainerHB fetch container HB
    FetchContainerHB() *ContainerHeartbeatReq
    // ResourceHBInterval fetch resource HB interface
    ResourceHBInterval() time.Duration
    // ContainerHBInterval fetch container HB interface
    ContainerHBInterval() time.Duration
    // HBHandler HB hander
    HBHandler() HeartbeatHandler
}

Adapter prophet adapter

type Cfg Uses

type Cfg struct {
    // RPCAddr prophet rpc addr
    RPCAddr string

    // MaxScheduleRetries maximum retry times for schedule
    MaxScheduleRetries int
    // MaxScheduleInterval maximum schedule interval per scheduler
    MaxScheduleInterval time.Duration
    // MinScheduleInterval minimum schedule interval per scheduler
    MinScheduleInterval time.Duration
    // TimeoutWaitOperatorComplete timeout for waitting teh operator complete
    TimeoutWaitOperatorComplete time.Duration
    // MaxFreezeScheduleInterval freeze the container for a while if shouldSchedule is returns false
    MaxFreezeScheduleInterval time.Duration
    // MaxAllowContainerDownDuration maximum down time of removed from replicas
    MaxAllowContainerDownDuration time.Duration
    // MaxRebalanceLeader maximum count of transfer leader operator
    MaxRebalanceLeader uint64
    // MaxRebalanceReplica maximum count of remove|add replica operator
    MaxRebalanceReplica uint64
    // MaxScheduleReplica maximum count of schedule replica operator
    MaxScheduleReplica uint64
    // MaxLimitSnapshotsCount maximum count of node about snapshot
    MaxLimitSnapshotsCount uint64
    // CountResourceReplicas replica number per resource
    CountResourceReplicas int
    // MinAvailableStorageUsedRate minimum storage used rate of container, if the rate is over this value, skip the container
    MinAvailableStorageUsedRate int
    // LocationLabels the label used for location
    LocationLabels []string
    // MaxRPCCons rpc conns
    MaxRPCCons int
    // MaxRPCConnIdle rpc conn max idle time
    MaxRPCConnIdle time.Duration
    // MaxRPCTimeout rpc max timeout
    MaxRPCTimeout time.Duration

    StorageNode bool
    LeaseTTL    int64
    Schedulers  []Scheduler
    Handler     RoleChangeHandler
}

Cfg prophet cfg

type ChangePeerType Uses

type ChangePeerType byte

ChangePeerType change peer type

type Container Uses

type Container interface {
    Serializable

    SetID(id uint64)
    ID() uint64
    Labels() []Pair
    State() State
    Clone() Container

    Marshal() ([]byte, error)
    Unmarshal(data []byte) error
}

Container is the resource container, the resource is running on the container the container is usually a node

type ContainerHeartbeatReq Uses

type ContainerHeartbeatReq struct {
    Data               []byte    `json:"data"`
    StorageCapacity    uint64    `json:"storageCapacity"`
    StorageAvailable   uint64    `json:"storageAvailable"`
    LeaderCount        uint64    `json:"leaderCount"`
    ReplicaCount       uint64    `json:"replicaCount"`
    SendingSnapCount   uint64    `json:"sendingSnapCount"`
    ReceivingSnapCount uint64    `json:"receivingSnapCount"`
    ApplyingSnapCount  uint64    `json:"applyingSnapCount"`
    Busy               bool      `json:"busy"`  // stop replica balance
    Block              bool      `json:"block"` // stop leader balance
    Container          Container `json:"-"`
}

ContainerHeartbeatReq container hb msg

func (*ContainerHeartbeatReq) Init Uses

func (req *ContainerHeartbeatReq) Init(adapter Adapter) error

Init init

func (*ContainerHeartbeatReq) Prepare Uses

func (req *ContainerHeartbeatReq) Prepare() error

Prepare prepare

type ContainerRuntime Uses

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

ContainerRuntime is the container runtime info

func (*ContainerRuntime) Clone Uses

func (cr *ContainerRuntime) Clone() *ContainerRuntime

Clone returns the container clone info

func (*ContainerRuntime) Downtime Uses

func (cr *ContainerRuntime) Downtime() time.Duration

Downtime returns the container down time

func (*ContainerRuntime) GetLabelValue Uses

func (cr *ContainerRuntime) GetLabelValue(key string) string

GetLabelValue returns label value of key

func (*ContainerRuntime) GetLocationID Uses

func (cr *ContainerRuntime) GetLocationID(keys []string) string

GetLocationID returns location id

func (*ContainerRuntime) HasNoneHeartbeat Uses

func (cr *ContainerRuntime) HasNoneHeartbeat() bool

HasNoneHeartbeat returns if received a heartbeat from the container

func (*ContainerRuntime) IsBlocked Uses

func (cr *ContainerRuntime) IsBlocked() bool

IsBlocked returns the container is blocked from balance if true.

func (*ContainerRuntime) IsUp Uses

func (cr *ContainerRuntime) IsUp() bool

IsUp returns the container is up state

func (*ContainerRuntime) LeaderScore Uses

func (cr *ContainerRuntime) LeaderScore() float64

LeaderScore returns score with leader

func (*ContainerRuntime) ReplicaScore Uses

func (cr *ContainerRuntime) ReplicaScore() float64

ReplicaScore returns score with replica

func (*ContainerRuntime) ResourceCount Uses

func (cr *ContainerRuntime) ResourceCount(kind ResourceKind) uint64

ResourceCount returns resource count by kind

func (*ContainerRuntime) ResourceScore Uses

func (cr *ContainerRuntime) ResourceScore(kind ResourceKind) float64

ResourceScore returns resource score by kind

func (*ContainerRuntime) StorageUsedBytes Uses

func (cr *ContainerRuntime) StorageUsedBytes() uint64

StorageUsedBytes returns container used storage with bytes

func (*ContainerRuntime) StorageUsedRatio Uses

func (cr *ContainerRuntime) StorageUsedRatio() int

StorageUsedRatio returns container used storage with rate

type Coordinator Uses

type Coordinator struct {
    sync.RWMutex
    // contains filtered or unexported fields
}

Coordinator resource coordinator

type Elector Uses

type Elector interface {
    // Stop stop elector
    Stop(group uint64)

    // CurrentLeader returns current leader
    CurrentLeader(group uint64) (string, error)

    // ElectionLoop run leader election loop
    ElectionLoop(ctx context.Context, group uint64, currentLeader string, becomeLeader, becomeFollower func())

    // ChangeLeaderTo change leader to
    ChangeLeaderTo(group uint64, oldLeader, newLeader string) error
}

Elector a leader elector

func NewElector Uses

func NewElector(client *clientv3.Client, options ...ElectorOption) (Elector, error)

NewElector create a elector

type ElectorOption Uses

type ElectorOption func(*electorOptions)

ElectorOption elector option

func WithLeaderLeaseSeconds Uses

func WithLeaderLeaseSeconds(value int64) ElectorOption

WithLeaderLeaseSeconds set leader lease in seconds

func WithLockIfBecomeLeader Uses

func WithLockIfBecomeLeader(value bool) ElectorOption

WithLockIfBecomeLeader set lock enable flag if become leader, If true, will add a distributed lock, and will unlock on become follower, ensure that the other nodes can be changed to leaders after the previous leader has processed the role changes.

func WithPrefix Uses

func WithPrefix(value string) ElectorOption

WithPrefix set data prefix in embed etcd server

type EmbeddedEtcdCfg Uses

type EmbeddedEtcdCfg struct {
    Name         string
    DataPath     string
    Join         string
    EmbedEtcdLog string

    URLsClient          string
    URLsAdvertiseClient string

    URLsPeer          string
    URLsAdvertisePeer string

    InitialCluster      string
    InitialClusterState string
}

EmbeddedEtcdCfg cfg for embedded etcd

type EventNotify Uses

type EventNotify struct {
    Event int    `json:"event"`
    Value []byte `json:"value"`
}

EventNotify event notify

func (*EventNotify) ReadInitEventValues Uses

func (evt *EventNotify) ReadInitEventValues(resourceF func([]byte, uint64), containerF func([]byte))

ReadInitEventValues read all resource info

func (*EventNotify) ReadLeaderChangerValue Uses

func (evt *EventNotify) ReadLeaderChangerValue() (uint64, uint64)

ReadLeaderChangerValue returns the target resource and the new leader returns resourceid, newleaderid

type Filter Uses

type Filter interface {
    // FilterSource returns true means skip this container for schedule
    FilterSource(container *ContainerRuntime) bool
    // FilterTarget returns true means skip this container for schedule
    FilterTarget(container *ContainerRuntime) bool
}

Filter is used for filter container

func NewBlockFilter Uses

func NewBlockFilter() Filter

NewBlockFilter returns a filter for choose resource container by block

func NewCacheFilter Uses

func NewCacheFilter(freezeCache *resourceFreezeCache) Filter

NewCacheFilter returns a filter for choose resource container by runtime cache

func NewDistinctScoreFilter Uses

func NewDistinctScoreFilter(cfg *Cfg, containers []*ContainerRuntime, source *ContainerRuntime) Filter

NewDistinctScoreFilter a filter for ensures that distinct score will not decrease.

func NewExcludedFilter Uses

func NewExcludedFilter(sources, targets map[uint64]struct{}) Filter

NewExcludedFilter returns a filter for choose resource container by excluded value

func NewHealthFilter Uses

func NewHealthFilter(cfg *Cfg) Filter

NewHealthFilter returns a filter for choose resource container by health info

func NewLabelFilter Uses

func NewLabelFilter(labels []Pair) Filter

NewLabelFilter returns label filter, if the container missing the labels, skip it.

func NewSnapshotCountFilter Uses

func NewSnapshotCountFilter(cfg *Cfg) Filter

NewSnapshotCountFilter returns snapshot filter

func NewStateFilter Uses

func NewStateFilter(cfg *Cfg) Filter

NewStateFilter returns a filter for choose resource container by state

func NewStorageThresholdFilter Uses

func NewStorageThresholdFilter(cfg *Cfg) Filter

NewStorageThresholdFilter returns a filter for choose resource container by storage rate

type Float64Data Uses

type Float64Data []float64

Float64Data is a named type for []float64 with helper methods

func (Float64Data) Get Uses

func (f Float64Data) Get(i int) float64

Get item in slice

func (Float64Data) Len Uses

func (f Float64Data) Len() int

Len returns length of slice

func (Float64Data) Less Uses

func (f Float64Data) Less(i, j int) bool

Less returns if one number is less than another

func (Float64Data) Max Uses

func (f Float64Data) Max() (float64, error)

Max returns the maximum number in the data

func (Float64Data) Mean Uses

func (f Float64Data) Mean() (float64, error)

Mean returns the mean of the data

func (Float64Data) Median Uses

func (f Float64Data) Median() (float64, error)

Median returns the median of the data

func (Float64Data) Min Uses

func (f Float64Data) Min() (float64, error)

Min returns the minimum number in the data

func (Float64Data) Mode Uses

func (f Float64Data) Mode() ([]float64, error)

Mode returns the mode of the data

func (Float64Data) Sum Uses

func (f Float64Data) Sum() (float64, error)

Sum returns the total of all the numbers in the data

func (Float64Data) Swap Uses

func (f Float64Data) Swap(i, j int)

Swap switches out two numbers in slice

type HeartbeatHandler Uses

type HeartbeatHandler interface {
    ChangeLeader(resourceID uint64, newLeader *Peer)
    ChangePeer(resourceID uint64, peer *Peer, changeType ChangePeerType)
}

HeartbeatHandler handle for heartbeat rsp

type InitWatcher Uses

type InitWatcher struct {
    Flag int `json:"flag"`
}

InitWatcher init watcher

type Job Uses

type Job struct {
    sync.RWMutex
    // contains filtered or unexported fields
}

Job is do for something with state

func (*Job) Cancel Uses

func (job *Job) Cancel()

Cancel cancel the job

func (*Job) GetResult Uses

func (job *Job) GetResult() interface{}

GetResult returns job result

func (*Job) IsCancelled Uses

func (job *Job) IsCancelled() bool

IsCancelled returns true if job state is Cancelled

func (*Job) IsCancelling Uses

func (job *Job) IsCancelling() bool

IsCancelling returns true if job state is Cancelling

func (*Job) IsComplete Uses

func (job *Job) IsComplete() bool

IsComplete return true means the job is complete.

func (*Job) IsFailed Uses

func (job *Job) IsFailed() bool

IsFailed returns true if job state is Failed

func (*Job) IsFinished Uses

func (job *Job) IsFinished() bool

IsFinished returns true if job state is Finished

func (*Job) IsNotComplete Uses

func (job *Job) IsNotComplete() bool

IsNotComplete return true means the job is not complete.

func (*Job) IsPending Uses

func (job *Job) IsPending() bool

IsPending returns true if job state is Pending

func (*Job) IsRunning Uses

func (job *Job) IsRunning() bool

IsRunning returns true if job state is Running

func (*Job) SetResult Uses

func (job *Job) SetResult(result interface{})

SetResult set result

type JobState Uses

type JobState int

JobState is the job state

type LocalStorage Uses

type LocalStorage interface {
    // Get returns the key value
    Get(key []byte) ([]byte, error)
    // Set sets the key value to the local storage
    Set(key, value []byte) error
    // Remove remove the key from the local storage
    Remove(key []byte) error
    // Range visit all values that start with prefix, set limit to 0 for no limit
    Range(prefix []byte, limit uint64, fn func(key, value []byte) bool) error
}

LocalStorage is the local data storage

type LocalStore Uses

type LocalStore interface {
    // BootstrapCluster bootstrap the cluster,
    BootstrapCluster(initResources ...Resource)

    // MustPutResource put the resource to local
    MustPutResource(Resource)

    // MustRemoveResource remove the res from the local
    MustRemoveResource(uint64)

    // MustAllocID returns the new id by pd
    MustAllocID() uint64

    // MustCountResources returns local resources count
    MustCountResources() int

    // MustLoadResources load all local resources
    MustLoadResources(func(value []byte) (uint64, error))
}

LocalStore the local data store

func NewLocalStore Uses

func NewLocalStore(meta Container, storage LocalStorage, pd Prophet) LocalStore

NewLocalStore returns a local store

type Logger Uses

type Logger interface {
    Info(v ...interface{})
    Infof(format string, v ...interface{})
    Debug(v ...interface{})
    Debugf(format string, v ...interface{})
    Warn(v ...interface{})
    Warnf(format string, v ...interface{})
    Error(v ...interface{})
    Errorf(format string, v ...interface{})
    Fatal(v ...interface{})
    Fatalf(format string, v ...interface{})
}

Logger logger

type Node Uses

type Node struct {
    Name string `json:"name"`
    Addr string `json:"addr"`
}

Node is prophet info

type Operator Uses

type Operator interface {
    ResourceID() uint64
    ResourceKind() ResourceKind
    Do(target *ResourceRuntime) (*resourceHeartbeatRsp, bool)
}

Operator is an interface to scheduler resource

type Option Uses

type Option func(*options)

Option is prophet create option

func ParseProphetOptions Uses

func ParseProphetOptions(name string) []Option

ParseProphetOptions parse the prophet options from command line parameter

func WithCountResourceReplicas Uses

func WithCountResourceReplicas(value int) Option

WithCountResourceReplicas replica number per resource

func WithEmbeddedEtcd Uses

func WithEmbeddedEtcd(cfg *EmbeddedEtcdCfg) Option

WithEmbeddedEtcd using embedded etcd cluster

func WithExternalEtcd Uses

func WithExternalEtcd(client *clientv3.Client) Option

WithExternalEtcd using external etcd cluster

func WithLeaseTTL Uses

func WithLeaseTTL(leaseTTL int64) Option

WithLeaseTTL prophet leader lease ttl

func WithLocationLabels Uses

func WithLocationLabels(value []string) Option

WithLocationLabels the label used for location

func WithMaxAllowContainerDownDuration Uses

func WithMaxAllowContainerDownDuration(value time.Duration) Option

WithMaxAllowContainerDownDuration maximum down time of removed from replicas

func WithMaxFreezeScheduleInterval Uses

func WithMaxFreezeScheduleInterval(value time.Duration) Option

WithMaxFreezeScheduleInterval freeze the container for a while if shouldSchedule is returns false

func WithMaxLimitSnapshotsCount Uses

func WithMaxLimitSnapshotsCount(value uint64) Option

WithMaxLimitSnapshotsCount maximum count of node about snapshot

func WithMaxRPCConnIdle Uses

func WithMaxRPCConnIdle(value time.Duration) Option

WithMaxRPCConnIdle set MaxRPCConnIdle

func WithMaxRPCCons Uses

func WithMaxRPCCons(value int) Option

WithMaxRPCCons set MaxRPCCons

func WithMaxRPCTimeout Uses

func WithMaxRPCTimeout(value time.Duration) Option

WithMaxRPCTimeout set MaxRPCTimeout

func WithMaxRebalanceLeader Uses

func WithMaxRebalanceLeader(value uint64) Option

WithMaxRebalanceLeader maximum count of transfer leader operator

func WithMaxRebalanceReplica Uses

func WithMaxRebalanceReplica(value uint64) Option

WithMaxRebalanceReplica maximum count of remove|add replica operator

func WithMaxScheduleInterval Uses

func WithMaxScheduleInterval(value time.Duration) Option

WithMaxScheduleInterval using MaxScheduleInterval maximum schedule interval per scheduler

func WithMaxScheduleReplica Uses

func WithMaxScheduleReplica(value uint64) Option

WithMaxScheduleReplica maximum count of schedule replica operator

func WithMaxScheduleRetries Uses

func WithMaxScheduleRetries(value int) Option

WithMaxScheduleRetries using MaxScheduleRetries maximum retry times for schedule

func WithMinAvailableStorageUsedRate Uses

func WithMinAvailableStorageUsedRate(value int) Option

WithMinAvailableStorageUsedRate minimum storage used rate of container, if the rate is over this value, skip the container

func WithMinScheduleInterval Uses

func WithMinScheduleInterval(value time.Duration) Option

WithMinScheduleInterval using MinScheduleInterval minimum schedule interval per scheduler

func WithRPCAddr Uses

func WithRPCAddr(value string) Option

WithRPCAddr set rpc address

func WithRoleChangeHandler Uses

func WithRoleChangeHandler(Handler RoleChangeHandler) Option

WithRoleChangeHandler using a role changed handler

func WithScheduler Uses

func WithScheduler(value Scheduler) Option

WithScheduler add a scheduler

func WithTimeoutWaitOperatorComplete Uses

func WithTimeoutWaitOperatorComplete(value time.Duration) Option

WithTimeoutWaitOperatorComplete timeout for waitting teh operator complete

type Pair Uses

type Pair struct {
    Key   string `json:"key"`
    Value string `json:"value"`
}

Pair key value pair

type Peer Uses

type Peer struct {
    ID          uint64 `json:"id"`
    ContainerID uint64 `json:"cid"`
}

Peer is the resource peer

func (*Peer) Clone Uses

func (p *Peer) Clone() *Peer

Clone returns a clone value

type PeerStats Uses

type PeerStats struct {
    Peer        *Peer
    DownSeconds uint64
}

PeerStats peer stats

func (*PeerStats) Clone Uses

func (ps *PeerStats) Clone() *PeerStats

Clone returns a clone value

type Prophet Uses

type Prophet interface {
    // Start start the prophet instance, this will start the lead election, heartbeat loop and listen requests
    Start()
    // GetStore returns the Store
    GetStore() Store
    // GetRPC returns the RPC client
    GetRPC() RPC
    // GetEtcdClient returns the internal etcd instance
    GetEtcdClient() *clientv3.Client
}

Prophet is the distributed scheduler and coordinator

func NewProphet Uses

func NewProphet(name string, adapter Adapter, opts ...Option) Prophet

NewProphet returns a prophet instance

type Queue Uses

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

Queue is the struct responsible for tracking the state of the queue.

func New Uses

func New(hint int64) *Queue

New is a constructor for a new threadsafe queue.

func (*Queue) Dispose Uses

func (q *Queue) Dispose() []interface{}

Dispose will dispose of this queue and returns the items disposed. Any subsequent calls to Get or Put will return an error.

func (*Queue) Disposed Uses

func (q *Queue) Disposed() bool

Disposed returns a bool indicating if this queue has had disposed called on it.

func (*Queue) Empty Uses

func (q *Queue) Empty() bool

Empty returns a bool indicating if this bool is empty.

func (*Queue) Get Uses

func (q *Queue) Get(number int64, items []interface{}) (int64, error)

Get retrieves items from the queue. If there are some items in the queue, get will return a number UP TO the number passed in as a parameter. If no items are in the queue, this method will pause until items are added to the queue.

func (*Queue) Len Uses

func (q *Queue) Len() int64

Len returns the number of items in this queue.

func (*Queue) Peek Uses

func (q *Queue) Peek() (interface{}, error)

Peek returns a the first item in the queue by value without modifying the queue.

func (*Queue) Poll Uses

func (q *Queue) Poll(number int64, items []interface{}, timeout time.Duration) (int64, error)

Poll retrieves items from the queue. If there are some items in the queue, Poll will return a number UP TO the number passed in as a parameter. If no items are in the queue, this method will pause until items are added to the queue or the provided timeout is reached. A non-positive timeout will block until items are added. If a timeout occurs, ErrTimeout is returned.

func (*Queue) Put Uses

func (q *Queue) Put(items ...interface{}) error

Put will add the specified items to the queue.

func (*Queue) PutOrUpdate Uses

func (q *Queue) PutOrUpdate(cmp func(interface{}, interface{}) bool, item interface{}) error

PutOrUpdate will add the specified item to the queue, update it if exists

type RPC Uses

type RPC interface {
    TiggerContainerHeartbeat()
    TiggerResourceHeartbeat(id uint64)
    AllocID() (uint64, error)
    AskSplit(res Resource) (uint64, []uint64, error)
}

RPC prophet rpc

type Resource Uses

type Resource interface {
    Serializable

    SetID(id uint64)
    ID() uint64
    Peers() []*Peer
    SetPeers(peers []*Peer)
    Stale(other Resource) bool
    Changed(other Resource) bool
    Clone() Resource
    Labels() []Pair

    Marshal() ([]byte, error)
    Unmarshal(data []byte) error
}

Resource resource

type ResourceHeartbeatReq Uses

type ResourceHeartbeatReq struct {
    Resource     Resource     `json:"-"`
    Data         []byte       `json:"data"`
    LeaderPeer   *Peer        `json:"leaderPeer"`
    DownPeers    []*PeerStats `json:"downPeers"`
    PendingPeers []*Peer      `json:"pendingPeers"`
}

ResourceHeartbeatReq resource hb msg

func (*ResourceHeartbeatReq) Init Uses

func (req *ResourceHeartbeatReq) Init(adapter Adapter) error

Init init

func (*ResourceHeartbeatReq) Prepare Uses

func (req *ResourceHeartbeatReq) Prepare() error

Prepare prepare

type ResourceKind Uses

type ResourceKind int

ResourceKind distinguishes different kinds of resources.

const (
    // LeaderKind leader
    LeaderKind ResourceKind = iota
    // ReplicaKind replica of resource
    ReplicaKind
)

type ResourceRuntime Uses

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

ResourceRuntime resource runtime info

func (*ResourceRuntime) Clone Uses

func (res *ResourceRuntime) Clone() *ResourceRuntime

Clone returns a clone resource runtime

func (*ResourceRuntime) GetContainerIDs Uses

func (res *ResourceRuntime) GetContainerIDs() map[uint64]struct{}

GetContainerIDs returns all container id

func (*ResourceRuntime) GetContainerPeer Uses

func (res *ResourceRuntime) GetContainerPeer(containerID uint64) *Peer

GetContainerPeer returns the peer in the container

func (*ResourceRuntime) GetPeer Uses

func (res *ResourceRuntime) GetPeer(peerID uint64) *Peer

GetPeer return the peer

func (*ResourceRuntime) GetPendingPeer Uses

func (res *ResourceRuntime) GetPendingPeer(peerID uint64) *Peer

GetPendingPeer returns pending peer

func (*ResourceRuntime) RemoveContainerPeer Uses

func (res *ResourceRuntime) RemoveContainerPeer(id uint64)

RemoveContainerPeer remove container peer

type RoleChangeHandler Uses

type RoleChangeHandler interface {
    BecomeLeader()
    BecomeFollower()
}

RoleChangeHandler prophet role change handler

type Runner Uses

type Runner struct {
    sync.RWMutex
    // contains filtered or unexported fields
}

Runner TODO

func NewRunner Uses

func NewRunner() *Runner

NewRunner returns a task runner

func (*Runner) AddNamedWorker Uses

func (s *Runner) AddNamedWorker(name string) (uint64, error)

AddNamedWorker add a named worker, the named worker has uniq queue, so jobs are linear execution

func (*Runner) IsNamedWorkerBusy Uses

func (s *Runner) IsNamedWorkerBusy(worker string) bool

IsNamedWorkerBusy returns true if named queue is not empty

func (*Runner) RunCancelableTask Uses

func (s *Runner) RunCancelableTask(task func(context.Context)) (uint64, error)

RunCancelableTask run a task that can be cancelled Example: err := s.RunCancelableTask(func(ctx context.Context) {

select {
case <-ctx.Done():
// cancelled
case <-time.After(time.Second):
	// do something
}

}) if err != nil {

// hanle error
return

}

func (*Runner) RunJob Uses

func (s *Runner) RunJob(desc string, task func() error) error

RunJob run a job

func (*Runner) RunJobWithNamedWorker Uses

func (s *Runner) RunJobWithNamedWorker(desc, worker string, task func() error) error

RunJobWithNamedWorker run a job in a named worker

func (*Runner) RunJobWithNamedWorkerWithCB Uses

func (s *Runner) RunJobWithNamedWorkerWithCB(desc, worker string, task func() error, cb func(*Job)) error

RunJobWithNamedWorkerWithCB run a job in a named worker

func (*Runner) RunTask Uses

func (s *Runner) RunTask(task func()) error

RunTask runs a task in new goroutine

func (*Runner) Stop Uses

func (s *Runner) Stop() error

Stop stop all task RunTask will failure with an error Wait complete for the tasks that already in execute Cancel the tasks that is not start

func (*Runner) StopCancelableTask Uses

func (s *Runner) StopCancelableTask(id uint64) error

StopCancelableTask stop cancelable spec task

type Runtime Uses

type Runtime struct {
    sync.RWMutex
    // contains filtered or unexported fields
}

Runtime runtime info

func (*Runtime) Container Uses

func (rc *Runtime) Container(id uint64) *ContainerRuntime

Container returns a cloned value of container runtime info

func (*Runtime) Containers Uses

func (rc *Runtime) Containers() []*ContainerRuntime

Containers returns the containers, using clone

func (*Runtime) RandFollowerResource Uses

func (rc *Runtime) RandFollowerResource(id uint64) *ResourceRuntime

RandFollowerResource returns the random follower resource

func (*Runtime) RandLeaderResource Uses

func (rc *Runtime) RandLeaderResource(id uint64) *ResourceRuntime

RandLeaderResource returns the random leader resource

func (*Runtime) Resource Uses

func (rc *Runtime) Resource(id uint64) *ResourceRuntime

Resource returns a cloned value of resource runtime info

func (*Runtime) ResourceContainers Uses

func (rc *Runtime) ResourceContainers(target *ResourceRuntime) []*ContainerRuntime

ResourceContainers returns containers that has the resource's peer

func (*Runtime) ResourceFollowerContainers Uses

func (rc *Runtime) ResourceFollowerContainers(res *ResourceRuntime) []*ContainerRuntime

ResourceFollowerContainers returns all containers for peers exclude leader

func (*Runtime) Resources Uses

func (rc *Runtime) Resources() []*ResourceRuntime

Resources returns the resources, using clone

type Scheduler Uses

type Scheduler interface {
    Name() string
    ResourceKind() ResourceKind
    ResourceLimit() uint64
    Prepare(rt *Runtime) error
    Cleanup(rt *Runtime)
    Schedule(rt *Runtime) Operator
}

Scheduler is an interface to schedule resources.

type Selector Uses

type Selector interface {
    SelectSource(containers []*ContainerRuntime, filters ...Filter) *ContainerRuntime
    SelectTarget(containers []*ContainerRuntime, filters ...Filter) *ContainerRuntime
}

Selector is an interface to select source and target container to schedule.

type Serializable Uses

type Serializable interface {
}

Serializable serializable

type State Uses

type State int

State is the state

const (
    // UP is normal state
    UP  State = iota
    // Down is the unavailable state
    Down
    // Tombstone is the destory state
    Tombstone
)

type Store Uses

type Store interface {
    // CampaignLeader is for leader election
    // if we are win the leader election, the enableLeaderFun will call
    CampaignLeader(ttl int64, enableLeaderFun, disableLeaderFun func()) error
    // ResignLeader delete leader itself and let others start a new election again.
    ResignLeader() error
    // GetCurrentLeader return current leader
    GetCurrentLeader() (*Node, error)
    // WatchLeader watch leader,
    // this funcation will return unitl the leader's lease is timeout
    // or server closed
    WatchLeader()

    // PutResource puts the meta to the store
    PutResource(meta Resource) error
    // PutContainer puts the meta to the store
    PutContainer(meta Container) error
    // GetContainer returns the spec container
    GetContainer(id uint64) (Container, error)
    // LoadResources load all resources
    LoadResources(limit int64, do func(Resource)) error
    // LoadContainers load all containers
    LoadContainers(limit int64, do func(Container)) error

    // AllocID returns the alloc id
    AllocID() (uint64, error)

    // PutBootstrapped put cluster is bootstrapped
    PutBootstrapped(container Container, resources ...Resource) (bool, error)
}

Store meta store

type Watcher Uses

type Watcher struct {
    sync.RWMutex
    // contains filtered or unexported fields
}

Watcher watcher client

func NewWatcher Uses

func NewWatcher(addrs ...string) *Watcher

NewWatcher returns a watcher for watch

func (*Watcher) Stop Uses

func (w *Watcher) Stop()

Stop stop watch

func (*Watcher) Watch Uses

func (w *Watcher) Watch(flag int) chan *EventNotify

Watch watch event

Package prophet imports 29 packages (graph). Updated 2019-08-20. Refresh now. Tools for package owners.