From c64f97082cc101cb828b039dae5f59e140b7c9e8 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Fri, 13 Jul 2018 11:04:05 +0300 Subject: [PATCH 1/3] Propagate statsd to config --- Gopkg.lock | 8 +++++++- Gopkg.toml | 4 ++++ config/config.go | 44 +++++++++++++++++++++++++++++++++++++++++++- main.go | 26 +++++++++++++++++++++++++- 4 files changed, 79 insertions(+), 3 deletions(-) diff --git a/Gopkg.lock b/Gopkg.lock index 18b8f74..c656a7f 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -100,9 +100,15 @@ 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 diff --git a/Gopkg.toml b/Gopkg.toml index f34b7d2..e61751d 100644 --- a/Gopkg.toml +++ b/Gopkg.toml @@ -52,3 +52,7 @@ [[constraint]] name = "github.com/beevik/ntp" version = "0.2.0" + +[[constraint]] + name = "gopkg.in/alexcesaro/statsd.v2" + version = "2.0.0" diff --git a/config/config.go b/config/config.go index aaab626..d595ee1 100644 --- a/config/config.go +++ b/config/config.go @@ -8,6 +8,7 @@ import ( "strings" "github.com/juju/errors" + statsd "gopkg.in/alexcesaro/statsd.v2" ) // Buffer sizes define internal socket buffer sizes. @@ -32,6 +33,14 @@ type Config struct { 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 } @@ -109,7 +118,9 @@ func NewConfig(debug, verbose bool, // nolint: gocyclo 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 @@ -174,5 +185,36 @@ func NewConfig(debug, verbose bool, // nolint: gocyclo 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 } diff --git a/main.go b/main.go index 7587613..da9b5fd 100644 --- a/main.go +++ b/main.go @@ -61,7 +61,7 @@ var ( 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"). @@ -72,6 +72,28 @@ var ( 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() ) @@ -96,6 +118,8 @@ func main() { *publicIPv6, *publicIPv6Port, *statsIP, *statsPort, *secret, *adtag, + *statsdIP, *statsdPort, *statsdNetwork, *statsdPrefix, + *statsdTagsFormat, *statsdTags, ) if err != nil { usage(err.Error()) From 5ddf3d77d16df29c1691a6dfbc72119593c69b69 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Fri, 13 Jul 2018 11:38:25 +0300 Subject: [PATCH 2/3] Implement integration with statsd --- main.go | 4 ++- stats/server.go | 20 ++++++++++--- stats/statsd.go | 79 +++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 98 insertions(+), 5 deletions(-) create mode 100644 stats/statsd.go 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 +} From 0344330230f9e07542a2eb6b127fad2330ba143c Mon Sep 17 00:00:00 2001 From: 9seconds Date: Fri, 13 Jul 2018 12:09:41 +0300 Subject: [PATCH 3/3] Document statsd integration --- README.md | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/README.md b/README.md index b65f5ac..08dd0eb 100644 --- a/README.md +++ b/README.md @@ -145,3 +145,33 @@ You will have this tool up and running on port 3128. Now curl 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`.