events

Documentation

Index

• Constants
• Variables
• func HandleOneExecutorEvent(name string, ch chan *eventtypes.StatusEvent, lstore *sync.Map, ...)
• func OnlyOne(ctx context.Context, lock *dlock.DLock) (func(), error)
• type EventLayer
• type EventMgr
  • func GetEventManager() *EventMgr
  • func (m *EventMgr) RegisterEventCallback(name string, cb executortypes.EventCbFn) error
  • func (m *EventMgr) UnRegisterEventCallback(name string)
• type GetExecutorJsonStoreFn
• type GetExecutorLocalStoreFn
• type InstanceStatus
• type InstancesWindow
• type LabelKey
• type Message
• type Notifier
  • func New(sender string, dest map[string]interface{}) (Notifier, error)
• type NotifierImpl
  • func (n *NotifierImpl) Send(content interface{}, options ...OpOperation) error
  • func (n *NotifierImpl) SendRaw(message *Message) error
• type Op
• type OpOperation
  • func WithDest(dest map[string]interface{}) OpOperation
  • func WithSender(sender string) OpOperation
• type RuntimeEvent
• type ServiceStatus
• type SetCallbackFn
• type WindowStatus

Constants

const (
	// status_update_event state
	KILLING     = "TASK_KILLING"
	KILLED      = "TASK_KILLED"
	RUNNING     = "TASK_RUNNING"
	FINISHED    = "TASK_FINISHED"
	FAILED      = "TASK_FAILED"
	STARTING    = "TASK_STARTING"
	STAGING     = "TASK_STAGING"
	DROPPED     = "TASK_DROPPED"
	UNKNOWN     = "TASK_UNKNOWN"
	UNREACHABLE = "TASK_UNREACHABLE"

	// instance state
	INSTANCE_RUNNING  = "Running"
	INSTANCE_FAILED   = "Failed"
	INSTANCE_FINISHED = "Finished"
	INSTANCE_KILLED   = "Killed"
	// instance_health_changed_event state after packaging
	// instance and service shared
	HEALTHY   = "Healthy"
	UNHEALTHY = "UnHealthy"

	WATCHED_DIR = "/dice/service/"

	// The suffix of the sender's name when eventbox is called, used to zone the stage of event distribution
	//The events sent by the initialize executor phase
	SUFFIX_INIT = "_INIT"
	// Events in the periodic compensation phase
	SUFFIX_PERIOD = "_PERIOD"
	// Events in other periods on ordinary time
	SUFFIX_NORMAL = "_NORMAL"
	// Temporarily assigned prefix for edas event
	SUFFIX_EDAS = "_EDAS"
	// EDASV2 Events in the periodic compensation phase
	SUFFIX_EDASV2_PERIOD = "_EDASV2_PERIOD"
	// EDASV2 Incremental event
	SUFFIX_EDASV2_NORMAL = "_EDASV2_NORMAL"
	// EDASV2 Initial event
	SUFFIX_EDASV2_INIT = "_EDASV2_INIT"
	// K8S Events in the periodic compensation phase
	SUFFIX_K8S_PERIOD = "_K8S_PERIOD"
	// K8S Incremental event
	SUFFIX_K8S_NORMAL = "_K8S_NORMAL"
	// K8S Initial event
	SUFFIX_K8S_INIT = "_K8S_INIT"

	// Event type
	// The calculated event corresponds to SUFFIX_NORMAL
	EVENTS_INCR = "increment"
	//Initialization or periodic compensation event，corresponds to SUFFIX_INIT和SUFFIX_PERIOD
	EVENTS_TOTAL = "total"

	// instance suffix, which is stored in lstore as part of the key, identifies the time when the instance starts the health check
	START_HC_TIME_SUFFIX = "_start_hc_time"

	// Determine the left window edge of the health check timeout, including the interval(15s) time
	LEFT_EDGE int64 = -20
	// Determine the right window edge of the health check timeout, consider that sigterm cannot kill the container, wait for a period of time (20s~30s) and then receive sigkill exit
	RIGHT_DEGE int64 = 40

	// marathon event type of instance state change
	STATUS_UPDATE_EVENT = "status_update_event"
	// marathon type of instance health change
	INSTANCE_HEALTH_CHANGED_EVENT = "instance_health_changed_event"

	// The scheduler instance change event is processed according to the above two types of events
	INSTANCE_STATUS = "instances-status"
)

Variables

var (
	EventboxDir = "/eventbox"
	MessageDir  = filepath.Join(EventboxDir, "messages")
)

Functions

func HandleOneExecutorEvent

func HandleOneExecutorEvent(name string, ch chan *eventtypes.StatusEvent, lstore *sync.Map, cb executortypes.EventCbFn, stopCh chan struct{})

HandleOneExecutorEvent Process the event logic of a single plug-in executor

func OnlyOne

func OnlyOne(ctx context.Context, lock *dlock.DLock) (func(), error)

Types

type EventLayer

type EventLayer struct {
	InstanceId  string
	ServiceName string
	RuntimeName string
}

type EventMgr

type EventMgr struct {

	// watch WATCHED_DIR path, synchronized to the local cache of the executor
	// 1, Used to calculate incremental state events
	// 2, Used for initialization and periodic compensation events
	MemEtcdStore jsonstore.JsonStore
	// contains filtered or unexported fields
}

func GetEventManager

func GetEventManager() *EventMgr

GetEventManager returns global instance

func (*EventMgr) RegisterEventCallback

func (m *EventMgr) RegisterEventCallback(name string, cb executortypes.EventCbFn) error

RegisterEventCallback Register the event handler function

func (*EventMgr) UnRegisterEventCallback

func (m *EventMgr) UnRegisterEventCallback(name string)

UnRegisterEventCallback Unregister the event handler function

type GetExecutorJsonStoreFn

type GetExecutorJsonStoreFn func(string) (*sync.Map, error)

type GetExecutorLocalStoreFn

type GetExecutorLocalStoreFn func(string) (*sync.Map, error)

type InstanceStatus

type InstanceStatus struct {
	ID             string `json:"id,omitempty"`
	Ip             string `json:"ip,omitempty"`
	InstanceStatus string `json:"instanceStatus,omitempty"`
	// stage records the exit stage of the container, which is only reflected in the exit (Killed, Failed, Finished) event in the incremental event
	// The current phases are:
	// a) Container startup phase (exit before health check timeout),"BeforeHealthCheckTimeout"
	// b) Health check timeout period (killed by health check）,"HealthCheckTimeout"
	// c) Post-health check stage (exit after the health check is completed),"AfterHealthCheckTimeout"
	Stage string `json:"stage,omitempty"`
	// Extended field
	Extra map[string]interface{} `json:"extra,omitempty"`
}

type InstancesWindow

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

type LabelKey

type LabelKey string

type Message

type Message struct {
	Sender  string                   `json:"sender"`
	Content interface{}              `json:"content"`
	Labels  map[LabelKey]interface{} `json:"labels"`
	Time    int64                    `json:"time,omitempty"` // UnixNano
	// contains filtered or unexported fields
}

type Notifier

type Notifier interface {
	Send(content interface{}, options ...OpOperation) error
	SendRaw(message *Message) error
}

func New

func New(sender string, dest map[string]interface{}) (Notifier, error)

type NotifierImpl

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

func (*NotifierImpl) Send

func (n *NotifierImpl) Send(content interface{}, options ...OpOperation) error

Write message to etcd

func (*NotifierImpl) SendRaw

func (n *NotifierImpl) SendRaw(message *Message) error

type Op

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

type OpOperation

type OpOperation func(*Op)

func WithDest

func WithDest(dest map[string]interface{}) OpOperation

func WithSender

func WithSender(sender string) OpOperation

type RuntimeEvent

type RuntimeEvent struct {
	RuntimeName     string          `json:"runtimeName"`
	ServiceStatuses []ServiceStatus `json:"serviceStatuses,omitempty"`
	// Temporary field, which identifies whether the corresponding runtime is deleted
	IsDeleted bool   `json:"isDeleted,omitempty"`
	EventType string `json:"eventType,omitempty"`
}

Structure stuffed into eventbox

type ServiceStatus

type ServiceStatus struct {
	ServiceName      string           `json:"serviceName"`
	ServiceStatus    string           `json:"serviceStatus,omitempty"`
	Replica          int              `json:"replica,omitempty"`
	InstanceStatuses []InstanceStatus `json:"instanceStatuses,omitempty"`

	HealthCheckDuration int `json:"healthCheckDuration,omitempty"`
}

type SetCallbackFn

type SetCallbackFn func(string) error

type WindowStatus

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