Merge pull request #27 from 9seconds/statsd

Integration with statsd

Gopkg.lock

   revision = "947dcec5ba9c011838740e680966fd7087a71d0d"
   version = "v2.2.6"
 
+[[projects]]
+  name = "gopkg.in/alexcesaro/statsd.v2"
+  packages = ["."]
+  revision = "7fea3f0d2fab1ad973e641e51dba45443a311a90"
+  version = "v2.0.0"
+
 [solve-meta]
   analyzer-name = "dep"
   analyzer-version = 1
-  inputs-digest = "f828340a30ea13c563829f9a37d0ff62974d4578411c9be02e61125dbdf98692"
+  inputs-digest = "7fad0f62feb7737b064d85cc4333a1a3e9298faec2afd864b4404f515fc7f17c"
   solver-name = "gps-cdcl"
   solver-version = 1

Gopkg.toml

 [[constraint]]
   name = "github.com/beevik/ntp"
   version = "0.2.0"
+
+[[constraint]]
+  name = "gopkg.in/alexcesaro/statsd.v2"
+  version = "2.0.0"

README.md

 port 3129 will show you some statistics if you are interested in.
 
 Also, you can use [run-mtg.sh](https://github.com/9seconds/mtg/blob/master/run-mtg.sh) script
+
+
+# statsd integration
+
+mtg provides an integration with statsd, you can enable it with command
+line interface. To enable it, you have to provide IP address of statsd
+service.
+
+Out of the box, mtg supports 2 additional dialects: [InfluxDB](https://www.influxdata.com/blog/getting-started-with-sending-statsd-metrics-to-telegraf-influxdb/)
+and [Datadog](https://docs.datadoghq.com/developers/dogstatsd/).
+
+All metrics are gauges. Here is the list of metrics and their meaning:
+
+| Metric name                     | Unit    | Description                                               |
+|---------------------------------|---------|-----------------------------------------------------------|
+| `connections.abridged.ipv4`     | number  | The number of active abridged IPv4 connections            |
+| `connections.abridged.ipv6`     | number  | The number of active abridged IPv6 connections            |
+| `connections.intermediate.ipv4` | number  | The number of active intermediate IPv4 connections        |
+| `connections.intermediate.ipv6` | number  | The number of active intermediate IPv6 connections        |
+| `connections.secure.ipv4`       | number  | The number of active secure intermediate IPv4 connections |
+| `connections.secure.ipv6`       | number  | The number of active secure intermediate IPv6 connections |
+| `crashes`                       | number  | An amount of crashes in client handlers                   |
+| `traffic.ingress`               | bytes   | Ingress traffic from the start of application (incoming)  |
+| `traffic.egress`                | bytes   | Egress traffic from the start of application (outgoing)   |
+| `speed.ingress`                 | bytes/s | Ingress bandwidth of the latest second (incoming traffic) |
+| `speed.egress`                  | bytes/s | Egress bandwidth of the latest second (outgoing traffic)  |
+
+All metrics are prefixed with given prefix. Default prefix is `mtg`.
+With such prefix metric name `traffic.ingress`, for example, would be
+`mtg.traffic.ingress`.

config/config.go

 	"strings"
 
 	"github.com/juju/errors"
+	statsd "gopkg.in/alexcesaro/statsd.v2"
 )
 
 // Buffer sizes define internal socket buffer sizes.
 	PublicIPv6 net.IP
 	StatsIP    net.IP
 
+	StatsD struct {
+		Addr       net.Addr
+		Prefix     string
+		Tags       map[string]string
+		TagsFormat statsd.TagFormat
+		Enabled    bool
+	}
+
 	Secret []byte
 	AdTag  []byte
 }
 	publicIPv4 net.IP, PublicIPv4Port uint16,
 	publicIPv6 net.IP, publicIPv6Port uint16,
 	statsIP net.IP, statsPort uint16,
-	secret, adtag string) (*Config, error) {
+	secret, adtag string,
+	statsdIP string, statsdPort uint16, statsdNetwork string, statsdPrefix string,
+	statsdTagsFormat string, statsdTags map[string]string) (*Config, error) {
 	secureMode := false
 	if strings.HasPrefix(secret, "dd") && len(secret) == 34 {
 		secureMode = true
 		SecureMode:     secureMode,
 	}
 
+	if statsdIP != "" {
+		conf.StatsD.Enabled = true
+		conf.StatsD.Prefix = statsdPrefix
+		conf.StatsD.Tags = statsdTags
+
+		var addr net.Addr
+		hostPort := net.JoinHostPort(statsdIP, strconv.Itoa(int(statsdPort)))
+		switch statsdNetwork {
+		case "tcp":
+			addr, err = net.ResolveTCPAddr("tcp", hostPort)
+		case "udp":
+			addr, err = net.ResolveUDPAddr("udp", hostPort)
+		default:
+			err = errors.Errorf("Unknown network %s", statsdNetwork)
+		}
+		if err != nil {
+			return nil, errors.Annotate(err, "Cannot resolve statsd address")
+		}
+		conf.StatsD.Addr = addr
+
+		switch statsdTagsFormat {
+		case "datadog":
+			conf.StatsD.TagsFormat = statsd.Datadog
+		case "influxdb":
+			conf.StatsD.TagsFormat = statsd.InfluxDB
+		case "":
+		default:
+			return nil, errors.Errorf("Unknown tags format %s", statsdTagsFormat)
+		}
+	}
+
 	return conf, nil
 }

main.go

 			Envar("MTG_IPV6_PORT").
 			Uint16()
 
-	statsIP = app.Flag("stats-ip", "Which IP bind stats server to").
+	statsIP = app.Flag("stats-ip", "Which IP bind stats server to.").
 		Short('t').
 		Envar("MTG_STATS_IP").
 		Default("127.0.0.1").
 			Default("3129").
 			Uint16()
 
+	statsdIP = app.Flag("statsd-ip", "Which IP should we use for working with statsd.").
+			Envar("MTG_STATSD_IP").
+			String()
+	statsdPort = app.Flag("statsd-port", "Which port should we use for working with statsd.").
+			Envar("MTG_STATSD_PORT").
+			Default("8125").
+			Uint16()
+	statsdNetwork = app.Flag("statsd-network", "Which network is used to work with statsd. Only 'tcp' and 'udp' are supported.").
+			Envar("MTG_STATSD_NETWORK").
+			Default("udp").
+			String()
+	statsdPrefix = app.Flag("statsd-prefix", "Which bucket prefix should we use for sending stats to statsd.").
+			Envar("MTG_STATSD_PREFIX").
+			Default("mtg").
+			String()
+	statsdTagsFormat = app.Flag("statsd-tags-format", "Which tag format should we use to send stats metrics. Valid options are 'datadog' and 'influxdb'.").
+				Envar("MTG_STATSD_TAGS_FORMAT").
+				String()
+	statsdTags = app.Flag("statsd-tags", "Tags to use for working with statsd (specified as 'key=value').").
+			Envar("MTG_STATSD_TAGS").
+			StringMap()
+
 	secret = app.Arg("secret", "Secret of this proxy.").Required().String()
 	adtag  = app.Arg("adtag", "ADTag of the proxy.").String()
 )
 		*publicIPv6, *publicIPv6Port,
 		*statsIP, *statsPort,
 		*secret, *adtag,
+		*statsdIP, *statsdPort, *statsdNetwork, *statsdPrefix,
+		*statsdTagsFormat, *statsdTags,
 	)
 	if err != nil {
 		usage(err.Error())
 		zap.S().Infow("Use direct connection to Telegram")
 	}
 
-	go stats.Start(conf)
+	if err := stats.Start(conf); err != nil {
+		panic(err)
+	}
 
 	server := proxy.NewProxy(conf)
 	if err := server.Serve(); err != nil {

stats/server.go

 var instance *stats
 
 // Start starts new statisitcs server.
-func Start(conf *config.Config) {
+func Start(conf *config.Config) error {
 	log := zap.S().Named("stats")
 
 	instance = &stats{
 		mutex:  &sync.RWMutex{},
 	}
 
+	if conf.StatsD.Enabled {
+		client, err := newStatsd(conf)
+		if err != nil {
+			return err
+		}
+		go client.run()
+	}
+
 	go crashManager()
 	go connectionManager()
 	go trafficManager()
 		}
 	})
 
-	if err := http.ListenAndServe(conf.StatAddr(), nil); err != nil {
-		log.Fatalw("Stats server has been stopped", "error", err)
-	}
+	go func() {
+		if err := http.ListenAndServe(conf.StatAddr(), nil); err != nil {
+			log.Fatalw("Stats server has been stopped", "error", err)
+		}
+	}()
+
+	return nil
 }

stats/statsd.go

+package stats
+
+import (
+	"time"
+
+	"github.com/juju/errors"
+	statsd "gopkg.in/alexcesaro/statsd.v2"
+
+	"github.com/9seconds/mtg/config"
+)
+
+const (
+	statsdConnectionsAbridgedV4 = "connections.abridged.ipv4"
+	statsdConnectionsAbridgedV6 = "connections.abridged.ipv6"
+
+	statsdConnectionsIntermediateV4 = "connections.intermediate.ipv4"
+	statsdConnectionsIntermediateV6 = "connections.intermediate.ipv6"
+
+	statsdConnectionsSecureV4 = "connections.secure.ipv4"
+	statsdConnectionsSecureV6 = "connections.secure.ipv6"
+
+	statsdTrafficIngress = "traffic.ingress"
+	statsdTrafficEgress  = "traffic.egress"
+
+	statsdSpeedIngress = "speed.ingress"
+	statsdSpeedEgress  = "speed.egress"
+
+	statsdCrashes = "crashes"
+)
+
+const statsdPollTime = time.Second
+
+type statsdExporter struct {
+	client *statsd.Client
+}
+
+func (s *statsdExporter) run() {
+	for range time.Tick(statsdPollTime) {
+		instance.mutex.Lock()
+
+		s.client.Gauge(statsdConnectionsAbridgedV4, instance.Connections.Abridged.IPv4)
+		s.client.Gauge(statsdConnectionsAbridgedV6, instance.Connections.Abridged.IPv6)
+		s.client.Gauge(statsdConnectionsIntermediateV4, instance.Connections.Intermediate.IPv4)
+		s.client.Gauge(statsdConnectionsIntermediateV6, instance.Connections.Intermediate.IPv6)
+		s.client.Gauge(statsdConnectionsSecureV4, instance.Connections.Secure.IPv4)
+		s.client.Gauge(statsdConnectionsSecureV6, instance.Connections.Secure.IPv6)
+		s.client.Gauge(statsdTrafficIngress, uint64(instance.Traffic.Ingress))
+		s.client.Gauge(statsdTrafficEgress, uint64(instance.Traffic.Egress))
+		s.client.Gauge(statsdSpeedIngress, uint64(instance.Speed.Ingress))
+		s.client.Gauge(statsdSpeedEgress, uint64(instance.Speed.Egress))
+		s.client.Gauge(statsdCrashes, instance.Crashes)
+
+		instance.mutex.Unlock()
+	}
+}
+
+func newStatsd(conf *config.Config) (*statsdExporter, error) {
+	options := []statsd.Option{
+		statsd.Network(conf.StatsD.Addr.Network()),
+		statsd.Address(conf.StatsD.Addr.String()),
+		statsd.Prefix(conf.StatsD.Prefix),
+	}
+
+	if conf.StatsD.TagsFormat > 0 {
+		options = append(options, statsd.TagsFormat(conf.StatsD.TagsFormat))
+		tags := make([]string, len(conf.StatsD.Tags)*2)
+		for k, v := range conf.StatsD.Tags {
+			tags = append(tags, k, v)
+		}
+		options = append(options, statsd.Tags(tags...))
+	}
+
+	client, err := statsd.New(options...)
+	if err != nil {
+		return nil, errors.Annotate(err, "Cannot create statsd client")
+	}
+
+	return &statsdExporter{client: client}, nil
+}