// Copyright 2013 Ooyala, Inc. /* Package statsd provides a Go dogstatsd client. Dogstatsd extends the popular statsd, adding tags and histograms and pushing upstream to Datadog. Refer to http://docs.datadoghq.com/guides/dogstatsd/ for information about DogStatsD. Example Usage: // Create the client c, err := statsd.New("127.0.0.1:8125") if err != nil { log.Fatal(err) } // Prefix every metric with the app name c.Namespace = "flubber." // Send the EC2 availability zone as a tag with every metric c.Tags = append(c.Tags, "us-east-1a") err = c.Gauge("request.duration", 1.2, nil, 1) statsd is based on go-statsd-client. */ package statsd import ( "bytes" "fmt" "math/rand" "net" "strconv" "strings" "sync" "time" ) // A Client is a handle for sending udp messages to dogstatsd. It is safe to // use one Client from multiple goroutines simultaneously. type Client struct { conn net.Conn // Namespace to prepend to all statsd calls Namespace string // Tags are global tags to be added to every statsd call Tags []string // BufferLength is the length of the buffer in commands. bufferLength int flushTime time.Duration commands []string stop bool sync.Mutex } // New returns a pointer to a new Client given an addr in the format "hostname:port". func New(addr string) (*Client, error) { udpAddr, err := net.ResolveUDPAddr("udp", addr) if err != nil { return nil, err } conn, err := net.DialUDP("udp", nil, udpAddr) if err != nil { return nil, err } client := &Client{conn: conn} return client, nil } // NewBuffered returns a Client that buffers its output and sends it in chunks. // Buflen is the length of the buffer in number of commands. func NewBuffered(addr string, buflen int) (*Client, error) { client, err := New(addr) if err != nil { return nil, err } client.bufferLength = buflen client.commands = make([]string, 0, buflen) client.flushTime = time.Millisecond * 100 go client.watch() return client, nil } // format a message from its name, value, tags and rate. Also adds global // namespace and tags. func (c *Client) format(name, value string, tags []string, rate float64) string { var buf bytes.Buffer if c.Namespace != "" { buf.WriteString(c.Namespace) } buf.WriteString(name) buf.WriteString(":") buf.WriteString(value) if rate < 1 { buf.WriteString(`|@`) buf.WriteString(strconv.FormatFloat(rate, 'f', -1, 64)) } tags = append(c.Tags, tags...) if len(tags) > 0 { buf.WriteString("|#") buf.WriteString(tags[0]) for _, tag := range tags[1:] { buf.WriteString(",") buf.WriteString(tag) } } return buf.String() } func (c *Client) watch() { for _ = range time.Tick(c.flushTime) { if c.stop { return } c.Lock() if len(c.commands) > 0 { // FIXME: eating error here c.flush() } c.Unlock() } } func (c *Client) append(cmd string) error { c.Lock() c.commands = append(c.commands, cmd) // if we should flush, lets do it if len(c.commands) == c.bufferLength { if err := c.flush(); err != nil { c.Unlock() return err } } c.Unlock() return nil } // flush the commands in the buffer. Lock must be held by caller. func (c *Client) flush() error { data := strings.Join(c.commands, "\n") _, err := c.conn.Write([]byte(data)) // clear the slice with a slice op, doesn't realloc c.commands = c.commands[:0] return err } func (c *Client) sendMsg(msg string) error { // if this client is buffered, then we'll just append this if c.bufferLength > 0 { return c.append(msg) } c.Lock() _, err := c.conn.Write([]byte(msg)) c.Unlock() return err } // send handles sampling and sends the message over UDP. It also adds global namespace prefixes and tags. func (c *Client) send(name, value string, tags []string, rate float64) error { if c == nil { return nil } if rate < 1 && rand.Float64() > rate { return nil } data := c.format(name, value, tags, rate) return c.sendMsg(data) } // Gauge measures the value of a metric at a particular time. func (c *Client) Gauge(name string, value float64, tags []string, rate float64) error { stat := fmt.Sprintf("%f|g", value) return c.send(name, stat, tags, rate) } // Count tracks how many times something happened per second. func (c *Client) Count(name string, value int64, tags []string, rate float64) error { stat := fmt.Sprintf("%d|c", value) return c.send(name, stat, tags, rate) } // Histogram tracks the statistical distribution of a set of values. func (c *Client) Histogram(name string, value float64, tags []string, rate float64) error { stat := fmt.Sprintf("%f|h", value) return c.send(name, stat, tags, rate) } // Set counts the number of unique elements in a group. func (c *Client) Set(name string, value string, tags []string, rate float64) error { stat := fmt.Sprintf("%s|s", value) return c.send(name, stat, tags, rate) } // TimeInMilliseconds sends timing information in milliseconds. // It is flushed by statsd with percentiles, mean and other info (https://github.com/etsy/statsd/blob/master/docs/metric_types.md#timing) func (c *Client) TimeInMilliseconds(name string, value float64, tags []string, rate float64) error { stat := fmt.Sprintf("%f|ms", value) return c.send(name, stat, tags, rate) } // Event sends the provided Event. func (c *Client) Event(e *Event) error { stat, err := e.Encode(c.Tags...) if err != nil { return err } return c.sendMsg(stat) } // SimpleEvent sends an event with the provided title and text. func (c *Client) SimpleEvent(title, text string) error { e := NewEvent(title, text) return c.Event(e) } // Close the client connection. func (c *Client) Close() error { if c == nil { return nil } c.stop = true return c.conn.Close() } // Events support type eventAlertType string const ( // Info is the "info" AlertType for events Info eventAlertType = "info" // Error is the "error" AlertType for events Error eventAlertType = "error" // Warning is the "warning" AlertType for events Warning eventAlertType = "warning" // Success is the "success" AlertType for events Success eventAlertType = "success" ) type eventPriority string const ( // Normal is the "normal" Priority for events Normal eventPriority = "normal" // Low is the "low" Priority for events Low eventPriority = "low" ) // An Event is an object that can be posted to your DataDog event stream. type Event struct { // Title of the event. Required. Title string // Text is the description of the event. Required. Text string // Timestamp is a timestamp for the event. If not provided, the dogstatsd // server will set this to the current time. Timestamp time.Time // Hostname for the event. Hostname string // AggregationKey groups this event with others of the same key. AggregationKey string // Priority of the event. Can be statsd.Low or statsd.Normal. Priority eventPriority // SourceTypeName is a source type for the event. SourceTypeName string // AlertType can be statsd.Info, statsd.Error, statsd.Warning, or statsd.Success. // If absent, the default value applied by the dogstatsd server is Info. AlertType eventAlertType // Tags for the event. Tags []string } // NewEvent creates a new event with the given title and text. Error checking // against these values is done at send-time, or upon running e.Check. func NewEvent(title, text string) *Event { return &Event{ Title: title, Text: text, } } // Check verifies that an event is valid. func (e Event) Check() error { if len(e.Title) == 0 { return fmt.Errorf("statsd.Event title is required") } if len(e.Text) == 0 { return fmt.Errorf("statsd.Event text is required") } return nil } // Encode returns the dogstatsd wire protocol representation for an event. // Tags may be passed which will be added to the encoded output but not to // the Event's list of tags, eg. for default tags. func (e Event) Encode(tags ...string) (string, error) { err := e.Check() if err != nil { return "", err } var buffer bytes.Buffer buffer.WriteString("_e{") buffer.WriteString(strconv.FormatInt(int64(len(e.Title)), 10)) buffer.WriteRune(',') buffer.WriteString(strconv.FormatInt(int64(len(e.Text)), 10)) buffer.WriteString("}:") buffer.WriteString(e.Title) buffer.WriteRune('|') buffer.WriteString(e.Text) if !e.Timestamp.IsZero() { buffer.WriteString("|d:") buffer.WriteString(strconv.FormatInt(int64(e.Timestamp.Unix()), 10)) } if len(e.Hostname) != 0 { buffer.WriteString("|h:") buffer.WriteString(e.Hostname) } if len(e.AggregationKey) != 0 { buffer.WriteString("|k:") buffer.WriteString(e.AggregationKey) } if len(e.Priority) != 0 { buffer.WriteString("|p:") buffer.WriteString(string(e.Priority)) } if len(e.SourceTypeName) != 0 { buffer.WriteString("|s:") buffer.WriteString(e.SourceTypeName) } if len(e.AlertType) != 0 { buffer.WriteString("|t:") buffer.WriteString(string(e.AlertType)) } if len(tags)+len(e.Tags) > 0 { all := make([]string, 0, len(tags)+len(e.Tags)) all = append(all, tags...) all = append(all, e.Tags...) buffer.WriteString("|#") buffer.WriteString(all[0]) for _, tag := range all[1:] { buffer.WriteString(",") buffer.WriteString(tag) } } return buffer.String(), nil }