Documentation
¶
Index ¶
- Constants
- type AppNames
- type DummySupervisor
- type EntityIDs
- type Option
- func WithAddress(address string) Option
- func WithBootStrapServers(bootStrapServers []string) Option
- func WithClusterName(clusterName string) Option
- func WithJoinSize(joinSize int) Option
- func WithLogEnabled(logEnabled bool) Option
- func WithLogLevels(logLevels map[string]logging.Level) Option
- func WithRebalance(enabled bool, intervalSeconds int, settleDelaySeconds int) Option
- func WithRebalanceEnabled(rebalanceEnabled bool) Option
- func WithRebalanceInterval(rebalanceInterval time.Duration) Option
- func WithRebalanceSettleDelay(rebalanceSettleDelay time.Duration) Option
- func WithReconciliationEnabled(reconciliationEnabled bool) Option
- func WithReconciliationOffset(reconciliationOffset int) Option
- func WithReplicaPoints(replicaPoints int) Option
- func WithStatsD(statsD bark.StatsReporter) Option
- type RecoverableEntity
- type Request
- type Response
- type Supervisor
- func (s *Supervisor) ActivateApp(app store.App)
- func (s *Supervisor) AppDetailsUpdateEventHandler(ctx json.Context, request *AppNames) (*Response, error)
- func (s *Supervisor) AppDetailsUpdateHandler(appName string)
- func (s *Supervisor) Boot()
- func (s *Supervisor) BootEntity(entity e.EntityInfo, forward bool) (err error)
- func (s *Supervisor) CloseRingPop()
- func (s *Supervisor) DeactivateApp(app store.App)
- func (s *Supervisor) HandleEvent(event events.Event)
- func (s *Supervisor) InitRingPop()
- func (s *Supervisor) OffloadOrPanic(nodeName string)
- func (s *Supervisor) RegisterHandler() error
- func (s *Supervisor) StartEntities(ctx json.Context, request *EntityIDs) (res *Response, err error)
- func (s *Supervisor) StartEntity(id string) (bool, error)
- func (s *Supervisor) StopEntities(ctx json.Context, request *EntityIDs) (res *Response, err error)
- func (s *Supervisor) StopEntity(id string) (bool, error)
- func (s *Supervisor) StopNode()
- func (s *Supervisor) TakeOverEntities(ctx json.Context, request *EntityIDs) (*Response, error)
- func (s *Supervisor) WaitForTermination()
- type SupervisorHandler
Constants ¶
const ( SUCCESS = iota // The request was successful. FAILED // The request failed. )
const ( StartEntities = "StartEntities" StopEntities = "StopEntities" AppDetailsUpdate = "AppDetailsUpdate" TakeOverEntities = "TakeOverEntities" )
const ( STOPPED = iota RUNNING )
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AppNames ¶
type AppNames struct {
Names []string
}
AppNames contains a slice of application names.
type DummySupervisor ¶
type DummySupervisor struct {
}
func (*DummySupervisor) ActivateApp ¶
func (d *DummySupervisor) ActivateApp(app store.App)
Implement if required
func (*DummySupervisor) BootEntity ¶
func (d *DummySupervisor) BootEntity(entity e.EntityInfo, bool bool) error
Implement if required
func (*DummySupervisor) DeactivateApp ¶
func (d *DummySupervisor) DeactivateApp(app store.App)
Implement if required
type Option ¶
type Option func(*options)
func WithAddress ¶
func WithBootStrapServers ¶
func WithClusterName ¶
func WithJoinSize ¶
func WithLogEnabled ¶
func WithRebalance ¶
WithRebalance applies rebalancing settings expressed in seconds, as read from configuration. Non-positive durations keep the defaults.
func WithRebalanceEnabled ¶
WithRebalanceEnabled turns on handing entities a node no longer owns back to their owner. When disabled the node still accepts entities handed to it by peers, so nodes can be switched over one at a time.
func WithRebalanceInterval ¶
WithRebalanceInterval sets how often a node hands entities it no longer owns back to their owner, independent of ring change events. Zero disables the periodic sweep; ring changes still trigger rebalancing.
func WithRebalanceSettleDelay ¶
WithRebalanceSettleDelay sets how long to wait after a ring change before rebalancing, so a burst of membership flaps results in a single sweep.
func WithReplicaPoints ¶
func WithStatsD ¶
func WithStatsD(statsD bark.StatsReporter) Option
type RecoverableEntity ¶
type RecoverableEntity struct {
Obj cluster_entity.Entity // The underlying entity.
Recovered int // The number of times the entity was recovered.
// contains filtered or unexported fields
}
RecoverableEntity is a wrapper for a recoverable entity.
func (RecoverableEntity) Init ¶
func (r RecoverableEntity) Init() error
Init initializes the recoverable entity before its asynchronous start.
func (RecoverableEntity) Start ¶
func (r RecoverableEntity) Start()
Start runs an initialized recoverable entity.
func (RecoverableEntity) Stop ¶
func (r RecoverableEntity) Stop()
Stop stops the recoverable entity.
type Request ¶
type Request struct {
// contains filtered or unexported fields
}
Request represents a request to be sent to a remote node.
type Response ¶
type Response struct {
ServerAddress string // The server address that sent the response.
Error string // The error message, if any.
Status int // The status of the response.
Accepted []string // Entity IDs the responding node took ownership of (TakeOverEntities only).
}
Response represents a response received from a remote node.
type Supervisor ¶
type Supervisor struct {
// contains filtered or unexported fields
}
noinspection SpellCheckingInspection
func NewSupervisor ¶
func NewSupervisor(entityFactory e.EntityFactory, clusterDao dao.ClusterDao, monitor p.Monitor, opt ...Option) *Supervisor
func (*Supervisor) ActivateApp ¶
func (s *Supervisor) ActivateApp(app store.App)
ActivateApp starts all the pollers for the app Forwards the pollers to start in case destination node is different TODO: Try to combine above methods
func (*Supervisor) AppDetailsUpdateEventHandler ¶
func (s *Supervisor) AppDetailsUpdateEventHandler(ctx json.Context, request *AppNames) (*Response, error)
AppDetailsUpdateEventHandler receives app update event Invalidates cache based on appName
func (*Supervisor) AppDetailsUpdateHandler ¶
func (s *Supervisor) AppDetailsUpdateHandler(appName string)
AppDetailsUpdateHandler invalidates in memory cache
func (*Supervisor) Boot ¶
func (s *Supervisor) Boot()
Boot fetch all the entities from DB and starts them one by one panics and stops the process in case there is any issue in starting any entity
func (*Supervisor) BootEntity ¶
func (s *Supervisor) BootEntity(entity e.EntityInfo, forward bool) (err error)
BootEntity perform following functions 1. Get all reachable members from a node 2. Check the destination node for a entity 3. Start/forward the entity depending on the destination node
func (*Supervisor) CloseRingPop ¶
func (s *Supervisor) CloseRingPop()
Stop ringpop, TChannel gracefully
func (*Supervisor) DeactivateApp ¶
func (s *Supervisor) DeactivateApp(app store.App)
DeactivateApp stops all the pollers for the app Forwards the pollers to stop in case destination node is different
func (*Supervisor) HandleEvent ¶
func (s *Supervisor) HandleEvent(event events.Event)
HandleEvent handle different events emitted by Ringpop TODO: Check all events
func (*Supervisor) InitRingPop ¶
func (s *Supervisor) InitRingPop()
initRingPop initializes ringpop with provided configurations
func (*Supervisor) OffloadOrPanic ¶
func (s *Supervisor) OffloadOrPanic(nodeName string)
Offload perform following functions 1. Get all entities assigned to the offloaded node 2. Check the destination node for a entity 3. Start the entity if the current node is the destination node 4. Reconcile schedules if there was any miss during offloading
func (*Supervisor) RegisterHandler ¶
func (s *Supervisor) RegisterHandler() error
RegisterHandler registers actions against respective methods
func (*Supervisor) StartEntities ¶
StartEntities iterates over the list of entityIds. Starts the entities on current node if destination node is same as own address or forward the entities to respective destination node
func (*Supervisor) StartEntity ¶
func (s *Supervisor) StartEntity(id string) (bool, error)
StartEntity starts an entity if it is not already running. Adds entity in in-memory concurrent map if the entity does not exist. Updates DB with entity status as Running.
func (*Supervisor) StopEntities ¶
StopEntities iterates over the list of entityIds. Sops the entities on current node if destination node is same as own address or forward the entities to respective destination node
func (*Supervisor) StopEntity ¶
func (s *Supervisor) StopEntity(id string) (bool, error)
StopEntity stops an entity if it is running. Removes entity from in-memory concurrent map. Updates DB with entity status as stopped.
func (*Supervisor) StopNode ¶
func (s *Supervisor) StopNode()
StopNode stops all the entities assigned to that node before it is brought down
func (*Supervisor) TakeOverEntities ¶
TakeOverEntities claims each entity that this node owns according to its own view of the ring and reports the claimed IDs back, so the caller only stops what this node is now running.
func (*Supervisor) WaitForTermination ¶
func (s *Supervisor) WaitForTermination()
WaitForTermination waits for OS signals to terminate Stops the node, closes stastDClient before exiting the program
type SupervisorHandler ¶
type SupervisorHandler interface {
// BootEntity boots an entity with the specified entity info and reconcile flag.
BootEntity(e.EntityInfo, bool) error
// DeactivateApp deactivates the specified application.
DeactivateApp(app store.App)
// ActivateApp activates the specified application.
ActivateApp(app store.App)
}
SupervisorHandler is an interface for a supervisor's handler.
Source Files