eventstream

Documentation

Index

• type Client
  • func NewEventStreamClient(d *yarpc.Dispatcher, clientName string, server string, ...) *Client
  • func NewLocalEventStreamClient(clientName string, handler *Handler, taskUpdateHandler EventHandler, ...) *Client
  • func (c *Client) Start()
  • func (c *Client) Stop()
• type ClientMetrics
  • func NewClientMetrics(scope tally.Scope) *ClientMetrics
• type EventHandler
• type Handler
  • func NewEventStreamHandler(bufferSize int, expectedClients []string, ...) *Handler
  • func (h *Handler) AddEvent(event *pb_eventstream.Event) error
  • func (h *Handler) GetEvents() ([]*pb_eventstream.Event, error)
  • func (h *Handler) InitStream(ctx context.Context, req *pb_eventstream.InitStreamRequest) (*pb_eventstream.InitStreamResponse, error)
  • func (h *Handler) WaitForEvents(ctx context.Context, req *pb_eventstream.WaitForEventsRequest) (*pb_eventstream.WaitForEventsResponse, error)
• type HandlerMetrics
  • func NewHandlerMetrics(scope tally.Scope) *HandlerMetrics
• type PurgedEventsProcessor

Types

type Client

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

Client is the event stream client

func NewEventStreamClient

func NewEventStreamClient(
	d *yarpc.Dispatcher,
	clientName string,
	server string,
	taskUpdateHandler EventHandler,
	parentScope tally.Scope,
) *Client

NewEventStreamClient creates a client that consumes from remote event stream handler

func NewLocalEventStreamClient

func NewLocalEventStreamClient(
	clientName string,
	handler *Handler,
	taskUpdateHandler EventHandler,
	parentScope tally.Scope,
) *Client

NewLocalEventStreamClient creates a local client that directly consumes from a local event stream handler

func (*Client) Start

func (c *Client) Start()

Start starts the client

func (*Client) Stop

func (c *Client) Stop()

Stop stops the client

type ClientMetrics

type ClientMetrics struct {
	EventsConsumed tally.Counter
	StreamIDChange tally.Counter
	PurgeOffset    tally.Gauge

	InitStreamAPI        tally.Counter
	InitStreamSuccess    tally.Counter
	InitStreamFail       tally.Counter
	WaitForEventsAPI     tally.Counter
	WaitForEventsSuccess tally.Counter
	WaitForEventsFailed  tally.Counter
}

ClientMetrics is the metrics for event stream client

func NewClientMetrics

func NewClientMetrics(scope tally.Scope) *ClientMetrics

NewClientMetrics creates a new ClientMetrics

type EventHandler

type EventHandler interface {
	// The event notification callback
	OnV0Event(event *pbeventstream.Event)

	// The events notification callback
	OnV0Events(events []*pbeventstream.Event)

	// Returns the event progress the handler has processed. The value
	// will be used by the client to determine the purgeOffset
	GetEventProgress() uint64
}

EventHandler is the interface for handling task update events

type Handler

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

Handler holds a circular buffer and serves request to pull data. This component is used in hostmgr and resmgr

func NewEventStreamHandler

func NewEventStreamHandler(
	bufferSize int,
	expectedClients []string,
	purgedEventsProcessor PurgedEventsProcessor,
	parentScope tally.Scope) *Handler

NewEventStreamHandler creates an EventStreamHandler

func (*Handler) AddEvent

func (h *Handler) AddEvent(event *pb_eventstream.Event) error

AddEvent adds a task Event or mesos status update into the inner circular buffer

func (*Handler) GetEvents

func (h *Handler) GetEvents() ([]*pb_eventstream.Event, error)

GetEvents returns all the events pending in circular buffer This method is primarily for debugging purpose

func (*Handler) InitStream

func (h *Handler) InitStream(
	ctx context.Context,
	req *pb_eventstream.InitStreamRequest) (*pb_eventstream.InitStreamResponse, error)

InitStream handles the initstream request

func (*Handler) WaitForEvents

func (h *Handler) WaitForEvents(
	ctx context.Context,
	req *pb_eventstream.WaitForEventsRequest) (*pb_eventstream.WaitForEventsResponse, error)

WaitForEvents handles the WaitForEvents request

type HandlerMetrics

type HandlerMetrics struct {
	Head     tally.Gauge
	Tail     tally.Gauge
	Size     tally.Gauge
	Capacity tally.Gauge

	UnexpectedClientError tally.Counter
	PurgeEventError       tally.Counter
	InvalidStreamIDError  tally.Counter

	AddEventAPI          tally.Counter
	AddEventSuccess      tally.Counter
	AddEventFail         tally.Counter
	AddEventDeDupe       tally.Counter
	InitStreamAPI        tally.Counter
	InitStreamSuccess    tally.Counter
	InitStreamFail       tally.Counter
	WaitForEventsAPI     tally.Counter
	WaitForEventsSuccess tally.Counter
	WaitForEventsFailed  tally.Counter
}

HandlerMetrics is the metrics for event stream handler

func NewHandlerMetrics

func NewHandlerMetrics(scope tally.Scope) *HandlerMetrics

NewHandlerMetrics creates a HandlerMetrics

type PurgedEventsProcessor

type PurgedEventsProcessor interface {
	EventPurged(events []*cirbuf.CircularBufferItem)
}

PurgedEventsProcessor is the interface to handle the purged data