consumer

Documentation

Index

• Constants
• Variables
• type Consumer
  • func New(trafficControllerUrl string, tlsConfig *tls.Config, ...) *Consumer
  • func (c *Consumer) Close() error
  • func (c *Consumer) ContainerEnvelopes(appGuid, authToken string) ([]*events.Envelope, error)
  • func (c *Consumer) ContainerMetrics(appGuid string, authToken string) ([]*events.ContainerMetric, error)
  • func (c *Consumer) FilteredFirehose(subscriptionId string, authToken string, filter EnvelopeFilter) (<-chan *events.Envelope, <-chan error)
  • func (c *Consumer) Firehose(subscriptionId string, authToken string) (<-chan *events.Envelope, <-chan error)
  • func (c *Consumer) FirehoseWithoutReconnect(subscriptionId string, authToken string) (<-chan *events.Envelope, <-chan error)
  • func (c *Consumer) RecentLogs(appGuid string, authToken string) ([]*events.LogMessage, error)
  • func (c *Consumer) RefreshTokenFrom(tr TokenRefresher)
  • func (c *Consumer) SetDebugPrinter(debugPrinter DebugPrinter)
  • func (c *Consumer) SetIdleTimeout(idleTimeout time.Duration)
  • func (c *Consumer) SetMaxRetryCount(count int)
  • func (c *Consumer) SetMaxRetryDelay(d time.Duration)
  • func (c *Consumer) SetMinRetryDelay(d time.Duration)
  • func (c *Consumer) SetOnConnectCallback(cb func())
  • func (c *Consumer) Stream(appGuid string, authToken string) (outputChan <-chan *events.Envelope, errorChan <-chan error)
  • func (c *Consumer) StreamWithoutReconnect(appGuid string, authToken string) (<-chan *events.Envelope, <-chan error)
  • func (c *Consumer) TailingLogs(appGuid, authToken string) (<-chan *events.LogMessage, <-chan error)
  • func (c *Consumer) TailingLogsWithoutReconnect(appGuid string, authToken string) (<-chan *events.LogMessage, <-chan error)
• type DebugPrinter
• type EnvelopeFilter
• type FirehoseOption
  • func WithEnvelopeFilter(filter EnvelopeFilter) FirehoseOption
  • func WithRetry(retry bool) FirehoseOption
• type TokenRefresher

Constants

const (
	DefaultMinRetryDelay = 500 * time.Millisecond
	DefaultMaxRetryDelay = time.Minute
	DefaultMaxRetryCount = 1000
)

Variables

var (
	// KeepAlive sets the interval between keep-alive messages sent by the client to loggregator.
	KeepAlive = 25 * time.Second

	ErrNotOK             = errors.New("unknown issue when making HTTP request to Loggregator")
	ErrNotFound          = ErrNotOK // NotFound isn't an accurate description of how this is used; please use ErrNotOK instead
	ErrBadResponse       = errors.New("bad server response")
	ErrBadRequest        = errors.New("bad client request")
	ErrLostConnection    = errors.New("remote server terminated connection unexpectedly")
	ErrMaxRetriesReached = errors.New("maximum number of connection retries reached")
)

Types

type Consumer

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

Consumer represents the actions that can be performed against trafficcontroller. See sync.go and async.go for trafficcontroller access methods.

func New

func New(trafficControllerUrl string, tlsConfig *tls.Config, proxy func(*http.Request) (*url.URL, error)) *Consumer

New creates a new consumer to a trafficcontroller.

func (*Consumer) Close

func (c *Consumer) Close() error

Close terminates all previously opened websocket connections to the traffic controller. It will return an error if there are no open connections, or if it has problems closing any connection.

func (*Consumer) ContainerEnvelopes

func (c *Consumer) ContainerEnvelopes(appGuid, authToken string) ([]*events.Envelope, error)

ContainerEnvelopes connects to trafficcontroller via its 'containermetrics' http(s) endpoint and returns the most recent dropsonde envelopes for an app.

func (*Consumer) ContainerMetrics

func (c *Consumer) ContainerMetrics(appGuid string, authToken string) ([]*events.ContainerMetric, error)

ContainerMetrics is deprecated in favor of ContainerEnvelopes, since returning the ContainerMetric type directly hides important information, like the timestamp.

The returned values will be the same as ContainerEnvelopes, just with the Envelope stripped out.

func (*Consumer) FilteredFirehose

func (c *Consumer) FilteredFirehose(
	subscriptionId string,
	authToken string,
	filter EnvelopeFilter,
) (<-chan *events.Envelope, <-chan error)

FilteredFirehose streams a filtered set of envelopes. It has functionality similar to Firehose.

func (*Consumer) Firehose

func (c *Consumer) Firehose(
	subscriptionId string,
	authToken string,
) (<-chan *events.Envelope, <-chan error)

Firehose streams all data. All clients with the same subscriptionId will receive a proportionate share of the message stream. Each pool of clients will receive the entire stream.

Messages are presented in the order received from the loggregator server. Chronological or other ordering is not guaranteed. It is the responsibility of the consumer of these channels to provide any desired sorting mechanism.

Whenever an error is encountered, the error will be sent down the error channel and Firehose will attempt to reconnect indefinitely.

func (*Consumer) FirehoseWithoutReconnect

func (c *Consumer) FirehoseWithoutReconnect(
	subscriptionId string,
	authToken string,
) (<-chan *events.Envelope, <-chan error)

FirehoseWithoutReconnect functions identically to Firehose but without any reconnect attempts when errors occur.

func (*Consumer) RecentLogs

func (c *Consumer) RecentLogs(appGuid string, authToken string) ([]*events.LogMessage, error)

RecentLogs connects to trafficcontroller via its 'recentlogs' http(s) endpoint and returns a slice of recent messages. It does not guarantee any order of the messages; they are in the order returned by trafficcontroller.

The noaa.SortRecent function is provided to sort the data returned by this method.

func (*Consumer) RefreshTokenFrom

func (c *Consumer) RefreshTokenFrom(tr TokenRefresher)

func (*Consumer) SetDebugPrinter

func (c *Consumer) SetDebugPrinter(debugPrinter DebugPrinter)

SetDebugPrinter sets the websocket connection to write debug information to debugPrinter.

func (*Consumer) SetIdleTimeout

func (c *Consumer) SetIdleTimeout(idleTimeout time.Duration)

func (*Consumer) SetMaxRetryCount

func (c *Consumer) SetMaxRetryCount(count int)

SetMaxRetryCount sets the maximum number of reconnnection attemps that methods on c (e.g. Firehose, Stream, TailingLogs) will make before failing.

Defaults to DefaultMaxRetryCount.

func (*Consumer) SetMaxRetryDelay

func (c *Consumer) SetMaxRetryDelay(d time.Duration)

SetMaxRetryDelay sets the maximum duration that automatically reconnecting methods on c (e.g. Firehose, Stream, TailingLogs) will sleep for after receiving many successive errors from the traffic controller.

Defaults to DefaultMaxRetryDelay.

func (*Consumer) SetMinRetryDelay

func (c *Consumer) SetMinRetryDelay(d time.Duration)

SetMinRetryDelay sets the duration that automatically reconnecting methods on c (e.g. Firehose, Stream, TailingLogs) will sleep for after receiving an error from the traffic controller.

Successive errors will double the sleep time, up to c's max retry delay, set by c.SetMaxRetryDelay.

Defaults to DefaultMinRetryDelay.

func (*Consumer) SetOnConnectCallback

func (c *Consumer) SetOnConnectCallback(cb func())

SetOnConnectCallback sets a callback function to be called with the websocket connection is established.

func (*Consumer) Stream

func (c *Consumer) Stream(appGuid string, authToken string) (outputChan <-chan *events.Envelope, errorChan <-chan error)

Stream listens indefinitely for all log and event messages.

Messages are presented in the order received from the loggregator server. Chronological or other ordering is not guaranteed. It is the responsibility of the consumer of these channels to provide any desired sorting mechanism.

Whenever an error is encountered, the error will be sent down the error channel and Stream will attempt to reconnect indefinitely.

func (*Consumer) StreamWithoutReconnect

func (c *Consumer) StreamWithoutReconnect(appGuid string, authToken string) (<-chan *events.Envelope, <-chan error)

StreamWithoutReconnect functions identically to Stream but without any reconnect attempts when errors occur.

func (*Consumer) TailingLogs

func (c *Consumer) TailingLogs(appGuid, authToken string) (<-chan *events.LogMessage, <-chan error)

TailingLogs listens indefinitely for log messages only; other event types are dropped. Whenever an error is encountered, the error will be sent down the error channel and TailingLogs will attempt to reconnect up to 5 times. After five failed reconnection attempts, TailingLogs will give up and close the error and LogMessage channels.

If c is closed, the returned channels will both be closed.

Errors must be drained from the returned error channel for it to continue retrying; if they are not drained, the connection attempts will hang.

func (*Consumer) TailingLogsWithoutReconnect

func (c *Consumer) TailingLogsWithoutReconnect(appGuid string, authToken string) (<-chan *events.LogMessage, <-chan error)

TailingLogsWithoutReconnect functions identically to TailingLogs but without any reconnect attempts when errors occur.

type DebugPrinter

type DebugPrinter interface {
	Print(title, dump string)
}

DebugPrinter is a type which handles printing debug information.

type EnvelopeFilter

type EnvelopeFilter int

const (
	LogMessages EnvelopeFilter = iota
	Metrics
)

type FirehoseOption

type FirehoseOption func(*firehose)

func WithEnvelopeFilter

func WithEnvelopeFilter(filter EnvelopeFilter) FirehoseOption

func WithRetry

func WithRetry(retry bool) FirehoseOption

type TokenRefresher

type TokenRefresher interface {
	RefreshAuthToken() (token string, authError error)
}