watchevent

Documentation

Index

• func InitWatchProcessor(cfg Config, watchEventMetric *metrics.Metrics)
• func NewWatchID(topic Topic) string
• func NewWatchProcessor(cfg Config, watchEventMetric *metrics.Metrics) *watchProcessor
• type Config
• type EventClient
• type StopSignal
  • func (s StopSignal) String() string
• type Topic
  • func GetTopicFromInput(topic string) Topic
  • func GetTopicFromTheEvent(event interface{}) Topic
• type WatchProcessor
  • func GetWatchProcessor() WatchProcessor

Functions

func InitWatchProcessor

func InitWatchProcessor(
	cfg Config,
	watchEventMetric *metrics.Metrics,
)

InitWatchProcessor initializes WatchProcessor singleton.

func NewWatchID

func NewWatchID(topic Topic) string

NewWatchID creates a new watch id UUID string for the specific watch client

func NewWatchProcessor

func NewWatchProcessor(
	cfg Config,
	watchEventMetric *metrics.Metrics,
) *watchProcessor

newWatchProcessor should only be used in unit tests. Call InitWatchProcessor for regular case use.

Types

type Config

type Config struct {
	// Size of per-client internal buffer
	BufferSize int `yaml:"buffer_size"`

	// Maximum number of concurrent watch clients
	MaxClient int `yaml:"max_client"`
}

Config for Watch API

type EventClient

type EventClient struct {
	Input  chan interface{}
	Signal chan StopSignal
}

EventClient represents a client which interested in task event changes.

type StopSignal

type StopSignal int

StopSignal is an event sent through event client Signal channel indicating a stop event for the specific watcher.

const (
	// StopSignalUnknown indicates a unspecified StopSignal.
	StopSignalUnknown StopSignal = iota
	// StopSignalCancel indicates the watch is cancelled by the user.
	StopSignalCancel
	// StopSignalOverflow indicates the watch is aborted due to event
	// overflow.
	StopSignalOverflow
)

func (StopSignal) String

func (s StopSignal) String() string

String returns a user-friendly name for the specific StopSignal

type Topic

type Topic string

Topic define the event object type processor supported

const (
	EventStream Topic = "eventstream"
	HostSummary Topic = "hostSummary"
	INVALID     Topic = ""
)

List of topic currently supported by watch processor

func GetTopicFromInput

func GetTopicFromInput(topic string) Topic

Map the string receive from input to a right topic

func GetTopicFromTheEvent

func GetTopicFromTheEvent(event interface{}) Topic

Retrieve the topic from the event object received during NotifyEventChange

type WatchProcessor

type WatchProcessor interface {
	// NewEventClient creates a new watch client for mesos task event changes.
	// Returns the watch id and a new instance of EventClient.
	NewEventClient(topic Topic) (string, *EventClient, error)

	// StopEventClients stops all the event clients on leadership change.
	StopEventClients()

	// StopEventClient stops a event watch client. Returns "not-found" error
	// if the corresponding watch client is not found.
	StopEventClient(watchID string) error

	// NotifyEventChange receives mesos task event, and notifies all the clients
	// which are interested in the event.
	NotifyEventChange(event interface{})
}

WatchProcessor interface is a central controller which handles watch client lifecycle, and task / job event fan-out.

func GetWatchProcessor

func GetWatchProcessor() WatchProcessor

GetWatchProcessor returns WatchProcessor singleton.