diff --git a/main.go b/main.go index da9b5fd..31f8c08 100644 --- a/main.go +++ b/main.go @@ -158,7 +158,9 @@ func main() { 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 { diff --git a/stats/server.go b/stats/server.go index a5369e4..8a111f9 100644 --- a/stats/server.go +++ b/stats/server.go @@ -14,7 +14,7 @@ import ( 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{ @@ -23,6 +23,14 @@ func Start(conf *config.Config) { 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() @@ -51,7 +59,11 @@ func Start(conf *config.Config) { } }) - 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 } diff --git a/stats/statsd.go b/stats/statsd.go new file mode 100644 index 0000000..2ecaa63 --- /dev/null +++ b/stats/statsd.go @@ -0,0 +1,79 @@ +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 +}