statsd

Documentation

Overview

Package statsd implements functionality for creating servers compatible with the statsd protocol. See https://github.com/etsy/statsd/blob/master/docs/metric_types.md for a description of the protocol.

The main components of the library are Receiver, Dispatcher, Aggregator and Flusher. Receiver is responsible for receiving metrics from the socket. Dispatcher dispatches received metrics among several Aggregators, which do aggregation based on type of the metric. At every FlushInterval Flusher flushes metrics via associated Backend objects.

Currently the library implements just a few types of Backend, one compatible with Graphite (http://graphite.wikidot.org), one for Datadog and one just for stdout, but any object implementing the Backend interface can be used with the library. See available backends at https://github.com/atlassian/gostatsd/tree/master/backend/backends.

As with the original etsy statsd, multiple backends can be used simultaneously.

Index

• Constants
• Variables
• func AddFlags(fs *pflag.FlagSet)
• type Aggregator
  • func NewAggregator(percentThresholds []float64, flushInterval, expiryInterval time.Duration, ...) Aggregator
• type AggregatorFactory
• type AggregatorFactoryFunc
  • func (f AggregatorFactoryFunc) Create() Aggregator
• type ConsoleServer
  • func (s *ConsoleServer) ListenAndServe(ctx context.Context) error
  • func (s *ConsoleServer) Serve(ctx context.Context, l net.Listener) error
• type Dispatcher
  • func NewDispatcher(numWorkers int, perWorkerBufferSize int, af AggregatorFactory) Dispatcher
• type Flusher
  • func NewFlusher(flushInterval time.Duration, dispatcher Dispatcher, receiver Receiver, ...) Flusher
• type FlusherStats
• type Handler
• type ProcessFunc
• type Receiver
  • func NewMetricReceiver(ns string, tags []string, cloud cloudTypes.Interface, handler Handler) Receiver
• type ReceiverStats
• type Server
  • func NewServer() *Server
  • func (s *Server) Run(ctx context.Context) error
  • func (s *Server) RunWithCustomSocket(ctx context.Context, sf SocketFactory) error
• type SocketFactory

Constants

const (
	// DefaultExpiryInterval is the default expiry interval for metrics.
	DefaultExpiryInterval = 5 * time.Minute
	// DefaultFlushInterval is the default metrics flush interval.
	DefaultFlushInterval = 1 * time.Second
	// DefaultMetricsAddr is the default address on which to listen for metrics.
	DefaultMetricsAddr = ":8125"
	// DefaultMaxQueueSize is the default maximum number of buffered metrics per worker.
	DefaultMaxQueueSize = 10000 // arbitrary
)

const (
	// ParamBackends is the name of parameter with backends.
	ParamBackends = "backends"
	// ParamConsoleAddr is the name of parameter with console address.
	ParamConsoleAddr = "console-addr"
	// ParamCloudProvider is the name of parameter with the name of cloud provider.
	ParamCloudProvider = "cloud-provider"
	// ParamDefaultTags is the name of parameter with the list of additional tags.
	ParamDefaultTags = "default-tags"
	// ParamExpiryInterval is the name of parameter with expiry interval for metrics.
	ParamExpiryInterval = "expiry-interval"
	// ParamFlushInterval is the name of parameter with metrics flush interval.
	ParamFlushInterval = "flush-interval"
	// ParamMaxReaders is the name of parameter with number of socket readers.
	ParamMaxReaders = "max-readers"
	// ParamMaxWorkers is the name of parameter with number of goroutines that aggregate metrics.
	ParamMaxWorkers = "max-workers"
	// ParamMaxQueueSize is the name of parameter with maximum number of buffered metrics per worker.
	ParamMaxQueueSize = "max-queue-size"
	// ParamMetricsAddr is the name of parameter with address on which to listen for metrics.
	ParamMetricsAddr = "metrics-addr"
	// ParamNamespace is the name of parameter with namespace for all metrics.
	ParamNamespace = "namespace"
	// ParamPercentThreshold is the name of parameter with list of applied percentiles.
	ParamPercentThreshold = "percent-threshold"
	// ParamWebAddr is the name of parameter with the address of the web-based console.
	ParamWebAddr = "web-addr"
)

const DefaultConsoleAddr = ":8126"

DefaultConsoleAddr is the default address on which a ConsoleServer will listen.

const DefaultWebConsoleAddr = ":8181"

DefaultWebConsoleAddr is the default address on which a WebConsoleServer will listen.

Variables

var DefaultBackends = []string{"graphite"}

DefaultBackends is the list of default backends.

var DefaultMaxReaders = runtime.NumCPU()

DefaultMaxReaders is the default number of socket reading goroutines.

var DefaultMaxWorkers = runtime.NumCPU()

DefaultMaxWorkers is the default number of goroutines that aggregate metrics.

var DefaultPercentThreshold = []string{"90"}

DefaultPercentThreshold is the default list of applied percentiles.

var DefaultTags = []string{""}

DefaultTags is the default list of additional tags.

Functions

func AddFlags

func AddFlags(fs *pflag.FlagSet)

AddFlags adds flags to the specified FlagSet.

Types

type Aggregator

type Aggregator interface {
	Receive(*types.Metric, time.Time)
	Flush(func() time.Time) *types.MetricMap
	Process(ProcessFunc)
	Reset(time.Time)
}

Aggregator is an object that aggregates statsd metrics. The function NewAggregator should be used to create the objects.

Incoming metrics should be passed via Receive function.

func NewAggregator

func NewAggregator(percentThresholds []float64, flushInterval, expiryInterval time.Duration, defaultTags []string) Aggregator

NewAggregator creates a new Aggregator object.

type AggregatorFactory

type AggregatorFactory interface {
	// Create creates Aggregator objects.
	Create() Aggregator
}

AggregatorFactory creates Aggregator objects.

type AggregatorFactoryFunc

type AggregatorFactoryFunc func() Aggregator

AggregatorFactoryFunc type is an adapter to allow the use of ordinary functions as AggregatorFactory.

func (AggregatorFactoryFunc) Create

func (f AggregatorFactoryFunc) Create() Aggregator

Create calls f().

type ConsoleServer

type ConsoleServer struct {
	Addr string
	Receiver
	Dispatcher
	Flusher
}

ConsoleServer is an object that listens for telnet connection on a TCP address Addr and provides a console interface to manage statsd server.

func (*ConsoleServer) ListenAndServe

func (s *ConsoleServer) ListenAndServe(ctx context.Context) error

ListenAndServe listens on the ConsoleServer's TCP network address and then calls Serve.

func (*ConsoleServer) Serve

func (s *ConsoleServer) Serve(ctx context.Context, l net.Listener) error

Serve accepts incoming connections on the listener and serves them a console interface to the Dispatcher and Receiver.

type Dispatcher

type Dispatcher interface {
	Run(context.Context) error
	DispatchMetric(context.Context, *types.Metric) error
	Flush(context.Context) <-chan *types.MetricMap
	Process(context.Context, ProcessFunc) *sync.WaitGroup
}

Dispatcher is responsible for managing Aggregators' lifecycle and dispatching metrics among them.

func NewDispatcher

func NewDispatcher(numWorkers int, perWorkerBufferSize int, af AggregatorFactory) Dispatcher

NewDispatcher creates a new Dispatcher with provided configuration.

type Flusher

type Flusher interface {
	Run(context.Context) error
	GetStats() FlusherStats
}

Flusher periodically flushes metrics from all Aggregators to Senders.

func NewFlusher

func NewFlusher(flushInterval time.Duration, dispatcher Dispatcher, receiver Receiver, defaultTags []string, backends []backendTypes.Backend) Flusher

NewFlusher creates a new Flusher with provided configuration.

type FlusherStats

type FlusherStats struct {
	LastFlush      time.Time // Last time the metrics where aggregated
	LastFlushError time.Time // Time of the last flush error
}

FlusherStats holds statistics about a Flusher.

type Handler

type Handler interface {
	DispatchMetric(context.Context, *types.Metric) error
	DispatchEvent(context.Context, *types.Event) error
}

Handler interface can be used to handle metrics and events for a Receiver.

type ProcessFunc

type ProcessFunc func(*types.MetricMap)

ProcessFunc is a function that gets executed by Aggregator with its state passed into the function.

type Receiver

type Receiver interface {
	Receive(context.Context, net.PacketConn) error
	GetStats() ReceiverStats
}

Receiver receives data on its PacketConn and converts lines into Metrics. For each types.Metric it calls Handler.HandleMetric()

func NewMetricReceiver

func NewMetricReceiver(ns string, tags []string, cloud cloudTypes.Interface, handler Handler) Receiver

NewMetricReceiver initialises a new Receiver.

type ReceiverStats

type ReceiverStats struct {
	LastPacket      time.Time
	BadLines        uint64
	PacketsReceived uint64
	MetricsReceived uint64
	EventsReceived  uint64
}

ReceiverStats holds statistics for a Receiver.

type Server

type Server struct {
	Backends         []string
	ConsoleAddr      string
	CloudProvider    string
	DefaultTags      []string
	ExpiryInterval   time.Duration
	FlushInterval    time.Duration
	MaxReaders       int
	MaxWorkers       int
	MaxQueueSize     int
	MaxMessengers    int
	MetricsAddr      string
	Namespace        string
	PercentThreshold []string
	WebConsoleAddr   string
	Viper            *viper.Viper
}

Server encapsulates all of the parameters necessary for starting up the statsd server. These can either be set via command line or directly.

func NewServer

func NewServer() *Server

NewServer will create a new Server with the default configuration.

func (*Server) Run

func (s *Server) Run(ctx context.Context) error

Run runs the server until context signals done.

func (*Server) RunWithCustomSocket

func (s *Server) RunWithCustomSocket(ctx context.Context, sf SocketFactory) error

RunWithCustomSocket runs the server until context signals done. Listening socket is created using sf.

type SocketFactory

type SocketFactory func() (net.PacketConn, error)

SocketFactory is an indirection layer over net.ListenPacket() to allow for different implementations.