nozzle

README

go-nozzle

go-nozzle is a pacakge for Go (golang) for building CloudFoundry(CF) nozzle. Nozzle is a program which consume data from the Loggregator Firehose and then select, buffer, and transform data, and forward it to other applications like Apache Kafka or external services like Data Dog.

With this package, you can create a nozzle client which gets the access token for firehose from UAA (when token is not provided), connects firehose endpoint and consumes logs (by default, it uses cloudfoundry/noaa client) and detects slow consumer alert. All you need to do is writing output part of nozzle (e.g., kafka producing).

Documentation is available on GoDoc.

Install

To install, use go get:

$ go get github.com/rakutentech/go-nozzle

Usage

The following is the simple example,

// Setup consumer configuration
config := &nozzle.Config{
	DopplerAddr:    "wss://doppler.cloudfoundry.net",
    UaaAddr:        "https://uaa.cloudfoundry.net",
    Username:       "tcnksm",
    Password:       "xyz",
    SubscriptionID: "go-nozzle-example-A",
}
   
// Create default consumer
consumer, _  := nozzle.NewDefaultConsumer(config)

// Consume events
event := <-consumer.Events()
... 

Also you can check the example usage of go-nozzle on example directory.

Author

Taichi Nakashima

Documentation

Overview

nozzle is a package for building your CloudFoundry(CF) nozzle. nozzle is a program which consume data from the Loggregator firehose (https://github.com/cloudfoundry/loggregator) and then select, buffer, and transform data and forward it to other applications, components or services.

This pacakge provides the consumer which (1) gets the access token for firehose, (2) connects firehose and consume logs, (3) detects slow consumer alert. To get starts, see Config and Consumer.

If you want to change the behavior of default consumer, then implement the interface of it.

Index

• type Config
• type Consumer
  • func NewDefaultConsumer(config *Config) (Consumer, error)
• type RawConsumer
• type SlowDetectCh
• type SlowDetector
• type TokenFetcher

Types

type Config

type Config struct {
	// DopplerAddr is a doppler firehose endpoint address to connect.
	// The address should start with 'wss://' (websocket endopint).
	DopplerAddr string

	// Token is an access token to connect to firehose. It's neccesary
	// to consume logs from doppler.
	//
	// If it's empty, the token is feched from UAA server.
	// To fetch token from UAA server, UaaAddr and Username/Password
	// for CF admin need to be set.
	Token string

	// SubscriptionID is unique id for a pool of clients of firehose.
	// For each SubscriptionID, all data will be distributed evenly
	// among that subscriber's client pool.
	SubscriptionID string

	// UaaAddr is UAA endpoint address. This is used for fetching access
	// token if Token is empty. To get token you also need to set
	// Username/Password for CloudFoundry admin.
	UaaAddr string

	// UaaTimeout is timeout to wait after sending request to uaa server.
	// The default value is 30 seconds.
	UaaTimeout time.Duration

	// Username is admin username of CloudFoundry. This is used for fetching
	// access token if Token is empty.
	Username string

	// Password is admin password of CloudFoundry. This is used for fetching
	// access token if Token is empty.
	Password string

	// Insecure is used for insecure connection with doppler.
	// Default value is false (Connect with TLS).
	Insecure bool

	// DebugPrinter is noaa.DebugPrinter. It's used for debugging
	// Noaa. Noaa is a client library to consume metric and log
	// messages from Doppler.
	DebugPrinter noaa.DebugPrinter

	// Logger is logger for go-nozzle. By default, output will be
	// discarded and not be displayed.
	Logger *log.Logger
	// contains filtered or unexported fields
}

Config is a configuration struct for go-nozzle. It contains all required values for using this pacakge. This is used for argument when constructing nozzle client.

type Consumer

type Consumer interface {
	// Events returns the read channel for the events that consumed by
	// rawConsumer(by default Noaa).
	Events() <-chan *events.Envelope

	// Detects returns the read channel that is notified slowConsumerAlerts
	// handled by SlowDetector.
	Detects() <-chan error

	// Error returns the read channel of erros that occured during consuming.
	Errors() <-chan error

	// Close stop consuming upstream events by RawConsumer and stop SlowDetector.
	Close() error
}

Consumer defines the interface of consumer it receives upstream firehose events and slowConsumerAlerts events and errors.

func NewDefaultConsumer

func NewDefaultConsumer(config *Config) (Consumer, error)

NewConsumer constructs a new consumer client for nozzle.

You need access token for consuming firehose log. There is 2 ways to construct. The one is to get token beforehand by yourself and use it. The other is to provide UAA endopoint with username/password for CloudFoundry admin to fetch the token.

It returns error if the token is empty or can not fetch token from UAA If token is not empty or successfully getting from UAA, then it starts to consume firehose events and detecting slowConsumerAlerts.

type RawConsumer

type RawConsumer interface {
	// Consume starts cosuming firehose events. It must return 2 channel.
	// The one is for sending the events from firehose
	// and the other is for error occured while consuming.
	// These channels are used donwstream process (SlowConsumer).
	Consume() (<-chan *events.Envelope, <-chan error)

	// Close closes connection with firehose. If any, returns error.
	Close() error
}

RawConsumer defines the interface for consuming events from doppler firehose. The events pulled by RawConsumer pass to slowDetector and check slowDetector.

By default, it uses https://github.com/cloudfoundry/noaa

type SlowDetectCh

type SlowDetectCh chan error

SlowDetectCh is channel used to send `slowConsumerAlert` event.

type SlowDetector

type SlowDetector interface {
	// Detect detects `slowConsumerAlert`. It works as pipe.
	// It receives events from upstream (RawConsumer) and inspects that events
	// and pass it to to downstream without modification.
	//
	// It returns SlowDetectCh and notify `slowConsumerAlert` there.
	Detect(<-chan *events.Envelope, <-chan error) (<-chan *events.Envelope, <-chan error, SlowDetectCh)

	// Stop stops slow consumer detection. If any returns error.
	Stop() error
}

SlowDetector defines the interface for detecting `slowConsumerAlert` event. By default, defaultSlowDetetor is used. It implements same detection logic as https://github.com/cloudfoundry-incubator/datadog-firehose-nozzle.

type TokenFetcher

type TokenFetcher interface {
	// Fetch fetches the token from Uaa and return it. If any, returns error.
	Fetch() (string, error)
}

TokenFetcher is the interface for fetching access token From UAA server. By default, defaultTokenFetcher (which is implemented with https://github.com/cloudfoundry-incubator/uaago) is used