statsd_exporter/main.go

324 lines
7.4 KiB
Go
Raw Normal View History

2013-07-02 16:16:39 +00:00
// Copyright (c) 2013, Prometheus Team
// All rights reserved.
//
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package main
import (
"flag"
"log"
"net"
"net/http"
"strconv"
"strings"
"sync"
"time"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/exp"
)
var listeningAddress = flag.String("listeningAddress", ":8080", "The address on which to expose generated Prometheus metrics.")
var statsdListeningAddress = flag.String("statsdListeningAddress", ":8126", "The UDP address on which to receive statsd metric lines.")
type CounterContainer struct {
sync.RWMutex
Elements map[string]prometheus.Counter
}
func NewCounterContainer() *CounterContainer {
return &CounterContainer{
Elements: make(map[string]prometheus.Counter),
}
}
func (c *CounterContainer) Get(metricName string) prometheus.Counter {
c.Lock()
defer c.Unlock()
counter, ok := c.Elements[metricName]
if !ok {
counter = prometheus.NewCounter()
c.Elements[metricName] = counter
prometheus.Register(metricName, "", prometheus.NilLabels, counter)
}
return counter
}
type GaugeContainer struct {
sync.RWMutex
Elements map[string]prometheus.Gauge
}
func NewGaugeContainer() *GaugeContainer {
return &GaugeContainer{
Elements: make(map[string]prometheus.Gauge),
}
}
func (c *GaugeContainer) Get(metricName string) prometheus.Gauge {
c.Lock()
defer c.Unlock()
gauge, ok := c.Elements[metricName]
if !ok {
gauge = prometheus.NewGauge()
c.Elements[metricName] = gauge
prometheus.Register(metricName, "", prometheus.NilLabels, gauge)
}
return gauge
}
type SummaryContainer struct {
sync.RWMutex
Elements map[string]prometheus.Histogram
ResetInterval time.Duration
}
func NewSummaryContainer() *SummaryContainer {
return &SummaryContainer{
Elements: make(map[string]prometheus.Histogram),
}
}
func (c *SummaryContainer) Get(metricName string) prometheus.Histogram {
c.Lock()
defer c.Unlock()
summary, ok := c.Elements[metricName]
if !ok {
summary = prometheus.NewDefaultHistogram()
c.Elements[metricName] = summary
prometheus.Register(metricName, "", prometheus.NilLabels, summary)
}
return summary
}
type Event interface {
MetricName() string
Value() float64
}
type CounterEvent struct {
metricName string
value float64
}
func (c *CounterEvent) MetricName() string { return c.metricName }
func (c *CounterEvent) Value() float64 { return c.value }
type GaugeEvent struct {
metricName string
value float64
}
func (g *GaugeEvent) MetricName() string { return g.metricName }
func (g *GaugeEvent) Value() float64 { return g.value }
type TimerEvent struct {
metricName string
value float64
}
func (t *TimerEvent) MetricName() string { return t.metricName }
func (t *TimerEvent) Value() float64 { return t.value }
type Events []Event
type Bridge struct {
Counters *CounterContainer
Gauges *GaugeContainer
Summaries *SummaryContainer
}
func escapeMetricName(metricName string) string {
// TODO: evaluate what kind of escaping we really want.
metricName = strings.Replace(metricName, "_", "__", -1)
metricName = strings.Replace(metricName, "-", "__", -1)
metricName = strings.Replace(metricName, ".", "_", -1)
return metricName
}
func (b *Bridge) Listen(e <-chan Events) {
for {
events := <-e
for _, event := range events {
metricName := escapeMetricName(event.MetricName())
switch event.(type) {
case *CounterEvent:
counter := b.Counters.Get(metricName + "_counter")
counter.IncrementBy(prometheus.NilLabels, event.Value())
case *GaugeEvent:
gauge := b.Gauges.Get(metricName + "_gauge")
gauge.Set(prometheus.NilLabels, event.Value())
case *TimerEvent:
summary := b.Summaries.Get(metricName + "_timer")
summary.Add(prometheus.NilLabels, event.Value())
default:
log.Println("Unsupported event type")
}
}
}
}
func NewBridge() *Bridge {
return &Bridge{
Counters: NewCounterContainer(),
Gauges: NewGaugeContainer(),
Summaries: NewSummaryContainer(),
}
}
type StatsDListener struct {
conn *net.UDPConn
}
func (l *StatsDListener) Listen(e chan<- Events) {
// TODO: evaluate proper size according to MTU
var buf [512]byte
for {
n, _, err := l.conn.ReadFromUDP(buf[0:])
if err != nil {
log.Fatal(err)
}
l.handlePacket(buf[0:n], e)
}
}
func (l *StatsDListener) handlePacket(packet []byte, e chan<- Events) {
lines := strings.Split(string(packet), "\n")
events := Events{}
for _, line := range lines {
if line == "" {
continue
}
elements := strings.Split(line, ":")
if len(elements) < 2 {
log.Println("Bad line from StatsD:", line)
continue
}
metric := elements[0]
samples := elements[1:]
for _, sample := range samples {
components := strings.Split(sample, "|")
samplingFactor := 1.0
if len(components) < 2 || len(components) > 3 {
log.Println("Bad component on line:", line)
continue
}
valueStr, statType := components[0], components[1]
value, err := strconv.Atoi(valueStr)
if err != nil {
log.Printf("Bad value %s on line: %s", valueStr, line)
continue
}
if len(components) == 3 {
if statType != "c" {
log.Println("Illegal sampling factor for non-counter metric on line", line)
}
samplingStr := components[2]
if samplingStr[0] != '@' {
log.Printf("Invalid sampling factor %s on line %s", samplingStr, line)
continue
}
samplingFactor, err = strconv.ParseFloat(samplingStr[1:], 64)
if err != nil {
log.Printf("Invalid sampling factor %s on line %s", samplingStr, line)
continue
}
if samplingFactor == 0 {
// This should never happen, but avoid division by zero if it does.
log.Printf("Invalid zero sampling factor %s on line %s, setting to 1", samplingStr, line)
samplingFactor = 1
}
}
var event Event
switch statType {
case "c":
event = &CounterEvent{
metricName: metric,
value: float64(value) / samplingFactor,
}
case "g":
event = &GaugeEvent{
metricName: metric,
value: float64(value),
}
case "ms":
event = &TimerEvent{
metricName: metric,
value: float64(value),
}
case "s":
log.Println("No support for StatsD sets in line", line)
default:
log.Printf("Bad stat type %s on line: %s", statType, line)
}
if event != nil {
events = append(events, event)
}
}
}
e <- events
}
func serveHTTP() {
exp.Handle(prometheus.ExpositionResource, prometheus.DefaultHandler)
http.ListenAndServe(*listeningAddress, exp.DefaultCoarseMux)
}
func udpAddrFromString(addr string) *net.UDPAddr {
host, portStr, err := net.SplitHostPort(*statsdListeningAddress)
if err != nil {
log.Fatal("Bad StatsD listening address", *statsdListeningAddress)
}
if host == "" {
host = "0.0.0.0"
}
ip, err := net.ResolveIPAddr("ip", host)
if err != nil {
log.Fatalf("Unable to resolve %s: %s", host, err)
}
port, err := strconv.Atoi(portStr)
if err != nil || port < 0 || port > 65535 {
log.Fatal("Bad port %s: %s", portStr, err)
}
return &net.UDPAddr{
IP: ip.IP,
Port: port,
Zone: ip.Zone,
}
}
func main() {
flag.Parse()
go serveHTTP()
events := make(chan Events, 1024)
defer close(events)
listenAddr := udpAddrFromString(*statsdListeningAddress)
conn, err := net.ListenUDP("udp", listenAddr)
if err != nil {
log.Fatal(err)
}
l := &StatsDListener{conn: conn}
go l.Listen(events)
bridge := NewBridge()
bridge.Listen(events)
}