From dfa53af567bff0beae03bc635095b375c7c07a7a Mon Sep 17 00:00:00 2001 From: 9seconds Date: Tue, 12 Nov 2019 12:13:00 +0300 Subject: [PATCH] Fix for statsd incorrect types --- README.md | 1 - config/config.go | 34 +++----- go.mod | 7 +- go.sum | 17 ++-- main.go | 6 -- stats/stats.go | 15 +--- stats/stats_prometheus.go | 33 +++----- stats/stats_statsd.go | 165 ++++++++++++++++++++++++++++---------- 8 files changed, 162 insertions(+), 116 deletions(-) diff --git a/README.md b/README.md index 22a5e9f..c6ca38a 100644 --- a/README.md +++ b/README.md @@ -181,7 +181,6 @@ supported environment variables: | `MTG_STATS_NAMESPACE` | `--stats-namespace` | `mtg` | Which namespace should be used for prometheus metrics. | | `MTG_STATSD_ADDR` | `--statsd-addr` | | IP:host addresses of statsd service. No defaults, by defaults we do not send anything there. | | `MTG_STATSD_PORT` | `--statsd-port` | `8125` | Which port should we use to work with statsd. | -| `MTG_STATSD_NETWORK` | `--statsd-network` | `udp` | Which protocol should we use to work with statsd. Possible options are `udp` and `tcp`. | | `MTG_STATSD_PREFIX` | `--statsd-prefix` | `mtg` | Which bucket prefix we should use. For example, if you set `mtg`, then metric `traffic.ingress` would be send as `mtg.traffic.ingress`. | | `MTG_STATSD_TAGS_FORMAT` | `--statsd-tags-format` | | Which tags format we should use. By default, we are using default vanilla statsd tags format but if you want to send directly to InfluxDB or Datadog, please specify it there. Possible options are `influxdb` and `datadog`. | | `MTG_STATSD_TAGS` | `--statsd-tags` | | Which tags should we send to statsd with our metrics. Please specify them as `key=value` pairs. | diff --git a/config/config.go b/config/config.go index 7db0e6e..31db1a9 100644 --- a/config/config.go +++ b/config/config.go @@ -9,8 +9,8 @@ import ( "net" "github.com/alecthomas/units" + statsd "github.com/smira/go-statsd" "go.uber.org/zap" - statsd "gopkg.in/alexcesaro/statsd.v2" ) type SecretMode uint8 @@ -47,7 +47,6 @@ const ( OptionTypeStatsBind OptionTypeStatsNamespace OptionTypeStatsdAddress - OptionTypeStatsdNetwork OptionTypeStatsdTagsFormat OptionTypeStatsdTags @@ -65,14 +64,14 @@ const ( ) type Config struct { - Bind *net.TCPAddr `json:"bind"` - PublicIPv4 *net.TCPAddr `json:"public_ipv4"` - PublicIPv6 *net.TCPAddr `json:"public_ipv6"` - StatsBind *net.TCPAddr `json:"stats_bind"` - StatsdAddr *net.TCPAddr `json:"stats_addr"` + Bind *net.TCPAddr `json:"bind"` + PublicIPv4 *net.TCPAddr `json:"public_ipv4"` + PublicIPv6 *net.TCPAddr `json:"public_ipv6"` + StatsBind *net.TCPAddr `json:"stats_bind"` + StatsdAddr *net.TCPAddr `json:"stats_addr"` + StatsdTagsFormat *statsd.TagFormat `json:"statsd_tags_format"` StatsNamespace string `json:"stats_namespace"` - StatsdNetwork string `json:"statsd_network"` CloakHost string `json:"cloak_host"` StatsdTags map[string]string `json:"statsd_tags"` @@ -84,10 +83,9 @@ type Config struct { MultiplexPerConnection int `json:"multiplex_per_connection"` - Debug bool `json:"debug"` - Verbose bool `json:"verbose"` - StatsdTagsFormat statsd.TagFormat `json:"statsd_tags_format"` - SecretMode SecretMode `json:"secret_mode"` + Debug bool `json:"debug"` + Verbose bool `json:"verbose"` + SecretMode SecretMode `json:"secret_mode"` Secret []byte `json:"secret"` AdTag []byte `json:"adtag"` @@ -125,21 +123,13 @@ func Init(options ...Opt) error { // nolint: gocyclo, funlen C.StatsNamespace = opt.Value.(string) case OptionTypeStatsdAddress: C.StatsdAddr = opt.Value.(*net.TCPAddr) - case OptionTypeStatsdNetwork: - value := opt.Value.(string) - switch value { - case "udp", "tcp": - C.StatsdNetwork = value - default: - return fmt.Errorf("unknown statsd network %v", value) - } case OptionTypeStatsdTagsFormat: value := opt.Value.(string) switch value { case "datadog": - C.StatsdTagsFormat = statsd.Datadog + C.StatsdTagsFormat = statsd.TagFormatDatadog case "influxdb": - C.StatsdTagsFormat = statsd.InfluxDB + C.StatsdTagsFormat = statsd.TagFormatInfluxDB default: return fmt.Errorf("incorrect statsd tag %s", value) } diff --git a/go.mod b/go.mod index 901f250..7b9155f 100644 --- a/go.mod +++ b/go.mod @@ -7,12 +7,13 @@ require ( github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d github.com/beevik/ntp v0.2.0 github.com/prometheus/client_golang v1.2.1 + github.com/prometheus/procfs v0.0.6 // indirect + github.com/smira/go-statsd v1.3.1 go.uber.org/multierr v1.4.0 // indirect go.uber.org/zap v1.12.0 - golang.org/x/crypto v0.0.0-20191108234033-bd318be0434a + golang.org/x/crypto v0.0.0-20191111213947-16651526fdb4 golang.org/x/net v0.0.0-20191109021931-daa7c04131f5 // indirect golang.org/x/sys v0.0.0-20191110163157-d32e6e3b99c4 - golang.org/x/tools v0.0.0-20191109212701-97ad0ed33101 // indirect + golang.org/x/tools v0.0.0-20191112005509-a3f652f18032 // indirect gopkg.in/alecthomas/kingpin.v2 v2.2.6 - gopkg.in/alexcesaro/statsd.v2 v2.0.0 ) diff --git a/go.sum b/go.sum index 3b89cb0..fb9d976 100644 --- a/go.sum +++ b/go.sum @@ -49,6 +49,8 @@ github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4= github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/google/go-cmp v0.3.0 h1:crn/baboCvb5fXaQ0IJ1SGTsTVrWpDsCWC8EGETZijY= github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= +github.com/google/go-cmp v0.3.1 h1:Xye71clBPdm5HgqGwUkwhbynsUJZhDbS20FvLhQ2izg= +github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI= github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU= @@ -95,9 +97,13 @@ github.com/prometheus/procfs v0.0.2 h1:6LJUbpNm42llc4HRCuvApCSWB/WfhuNo9K98Q9sNG github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA= github.com/prometheus/procfs v0.0.5 h1:3+auTFlqw+ZaQYJARz6ArODtkaIwtvBTx3N2NehQlL8= github.com/prometheus/procfs v0.0.5/go.mod h1:4A/X28fw3Fc593LaREMrKMqOKvUAntwMDaekg4FpcdQ= +github.com/prometheus/procfs v0.0.6 h1:0qbH+Yqu/cj1ViVLvEWCP6qMQ4efWUj6bQqOEA0V0U4= +github.com/prometheus/procfs v0.0.6/go.mod h1:7Qr8sr6344vo1JqZ6HhLceV9o3AJ1Ff+GxbHq6oeK9A= github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4= github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo= github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE= +github.com/smira/go-statsd v1.3.1 h1:JalGiHNdK7GqVAPpg7j0Kwp2jZrz/fCg/B4ZuNuBY2w= +github.com/smira/go-statsd v1.3.1/go.mod h1:1srXJ9/pbnN04G8f4F1jUzsGOnwkPKXciyqpewGlkC4= github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA= github.com/spaolacci/murmur3 v1.0.1-0.20190317074736-539464a789e9/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= @@ -121,8 +127,8 @@ golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnf golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2 h1:VklqNMn3ovrHsnt90PveolxSbWFaJdECFbxSq0Mqo2M= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= -golang.org/x/crypto v0.0.0-20191108234033-bd318be0434a h1:R/qVym5WAxsZWQqZCwDY/8sdVKV1m1WgU4/S5IRQAzc= -golang.org/x/crypto v0.0.0-20191108234033-bd318be0434a/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/crypto v0.0.0-20191111213947-16651526fdb4 h1:AGVXd+IAyeAb3FuQvYDYQ9+WR2JHm0+C0oYJaU1C4rs= +golang.org/x/crypto v0.0.0-20191111213947-16651526fdb4/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/lint v0.0.0-20190930215403-16217165b5de h1:5hukYrvBGR8/eNkX5mdUezrA6JiaEZDtJb9Ei+1LlBs= golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc= @@ -137,6 +143,7 @@ golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4 h1:YUO/7uOKsKeq9UokNS62b8FYywz3ker1l1vDZRCRefw= golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -152,13 +159,11 @@ golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgw golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5 h1:hKsoRgsbwY1NafxrwTs+k64bikrLBkAgPir1TNCj3Zs= golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= -golang.org/x/tools v0.0.0-20191109212701-97ad0ed33101 h1:LCmXVkvpQCDj724eX6irUTPCJP5GelFHxqGSWL2D1R0= -golang.org/x/tools v0.0.0-20191109212701-97ad0ed33101/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20191112005509-a3f652f18032 h1:Hp/Ke3YMUvqiVTvaoElioq98ROVmHsouhctlE8sVGpo= +golang.org/x/tools v0.0.0-20191112005509-a3f652f18032/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= -gopkg.in/alexcesaro/statsd.v2 v2.0.0 h1:FXkZSCZIH17vLCO5sO2UucTHsH9pc+17F6pl3JVCwMc= -gopkg.in/alexcesaro/statsd.v2 v2.0.0/go.mod h1:i0ubccKGzBVNBpdGV5MocxyA/XlLUJzA7SLonnE4drU= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/main.go b/main.go index 6864153..2d2e701 100644 --- a/main.go +++ b/main.go @@ -72,11 +72,6 @@ var ( "Host:port of statsd server"). Envar("MTG_STATSD_ADDR"). TCP() - runStatsdNetwork = runCommand.Flag("statsd-network", - "Which network is used to work with statsd. Only 'tcp' and 'udp' are supported."). - Envar("MTG_STATSD_NETWORK"). - Default("udp"). - Enum("udp", "tcp") runStatsdTagsFormat = runCommand.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"). @@ -139,7 +134,6 @@ func main() { config.Opt{Option: config.OptionTypeStatsBind, Value: *runStatsBind}, config.Opt{Option: config.OptionTypeStatsNamespace, Value: *runStatsNamespace}, config.Opt{Option: config.OptionTypeStatsdAddress, Value: *runStatsdAddress}, - config.Opt{Option: config.OptionTypeStatsdNetwork, Value: *runStatsdNetwork}, config.Opt{Option: config.OptionTypeStatsdTagsFormat, Value: *runStatsdTagsFormat}, config.Opt{Option: config.OptionTypeStatsdTags, Value: *runStatsdTags}, config.Opt{Option: config.OptionTypeWriteBufferSize, Value: *runWriteBufferSize}, diff --git a/stats/stats.go b/stats/stats.go index b4ccad7..d6b94af 100644 --- a/stats/stats.go +++ b/stats/stats.go @@ -14,20 +14,9 @@ var Stats Interface func Init(ctx context.Context) error { mux := http.NewServeMux() - instancePrometheus, err := newStatsPrometheus(mux) - if err != nil { - return fmt.Errorf("cannot initialize prometheus: %w", err) - } - - stats := []Interface{instancePrometheus} - + stats := []Interface{newStatsPrometheus(mux)} if config.C.StatsdAddr != nil { - instanceStatsd, err := newStatsStatsd() - if err != nil { - return fmt.Errorf("cannot inialize statsd: %w", err) - } - - stats = append(stats, instanceStatsd) + stats = append(stats, newStatsStatsd()) } listener, err := net.Listen("tcp", config.C.StatsBind.String()) diff --git a/stats/stats_prometheus.go b/stats/stats_prometheus.go index 509f864..7ec94c6 100644 --- a/stats/stats_prometheus.go +++ b/stats/stats_prometheus.go @@ -1,7 +1,6 @@ package stats import ( - "fmt" "net" "net/http" "strconv" @@ -17,7 +16,7 @@ type statsPrometheus struct { connections *prometheus.GaugeVec telegramConnections *prometheus.GaugeVec traffic *prometheus.GaugeVec - crashes prometheus.Gauge + crashes prometheus.Counter replayAttacks prometheus.Counter } @@ -88,7 +87,7 @@ func (s *statsPrometheus) ReplayDetected() { s.replayAttacks.Inc() } -func newStatsPrometheus(mux *http.ServeMux) (Interface, error) { +func newStatsPrometheus(mux *http.ServeMux) Interface { registry := prometheus.NewPedanticRegistry() instance := &statsPrometheus{ @@ -107,7 +106,7 @@ func newStatsPrometheus(mux *http.ServeMux) (Interface, error) { Name: "traffic", Help: "Traffic passed through the proxy in bytes.", }, []string{"direction"}), - crashes: prometheus.NewGauge(prometheus.GaugeOpts{ + crashes: prometheus.NewCounter(prometheus.CounterOpts{ Namespace: config.C.StatsNamespace, Name: "crashes", Help: "How many crashes happened.", @@ -119,28 +118,14 @@ func newStatsPrometheus(mux *http.ServeMux) (Interface, error) { }), } - if err := registry.Register(instance.connections); err != nil { - return nil, fmt.Errorf("cannot register metrics for connections: %w", err) - } - - if err := registry.Register(instance.telegramConnections); err != nil { - return nil, fmt.Errorf("cannot register metrics for telegram connections: %w", err) - } - - if err := registry.Register(instance.traffic); err != nil { - return nil, fmt.Errorf("cannot register metrics for traffic: %w", err) - } - - if err := registry.Register(instance.crashes); err != nil { - return nil, fmt.Errorf("cannot register metrics for crashes: %w", err) - } - - if err := registry.Register(instance.replayAttacks); err != nil { - return nil, fmt.Errorf("cannot register metrics for replays: %w", err) - } + registry.MustRegister(instance.connections) + registry.MustRegister(instance.telegramConnections) + registry.MustRegister(instance.traffic) + registry.MustRegister(instance.crashes) + registry.MustRegister(instance.replayAttacks) handler := promhttp.HandlerFor(registry, promhttp.HandlerOpts{}) mux.Handle("/", handler) - return instance, nil + return instance } diff --git a/stats/stats_statsd.go b/stats/stats_statsd.go index c8b8feb..19ed1b7 100644 --- a/stats/stats_statsd.go +++ b/stats/stats_statsd.go @@ -5,23 +5,74 @@ import ( "net" "strconv" "strings" + "sync" + "time" - "gopkg.in/alexcesaro/statsd.v2" + statsd "github.com/smira/go-statsd" + "go.uber.org/zap" "mtg/config" "mtg/conntypes" ) +var ( + tagTrafficIngress = &statsStatsdTag{ + name: "ingress", + tag: statsd.StringTag("type", "ingress"), + } + tagTrafficEgress = &statsStatsdTag{ + name: "egress", + tag: statsd.StringTag("type", "egress"), + } + + tagConnectionTypeAbridged = &statsStatsdTag{ + name: "abridged", + tag: statsd.StringTag("type", "abridged"), + } + tagConnectionTypeIntermediate = &statsStatsdTag{ + name: "intermediate", + tag: statsd.StringTag("type", "intermediate"), + } + tagConnectionTypeSecured = &statsStatsdTag{ + name: "secured", + tag: statsd.StringTag("type", "secured"), + } + + tagConnectionProtocol4 = &statsStatsdTag{ + name: "ipv4", + tag: statsd.StringTag("protocol", "ipv4"), + } + tagConnectionProtocol6 = &statsStatsdTag{ + name: "ipv6", + tag: statsd.StringTag("protocol", "ipv6"), + } +) + +type statsStatsdTag struct { + tag statsd.Tag + name string +} + +type statsStatsdLogger struct { + log *zap.SugaredLogger +} + +func (s *statsStatsdLogger) Printf(msg string, args ...interface{}) { + s.log.Debugw(fmt.Sprintf(msg, args...)) +} + type statsStatsd struct { - client *statsd.Client + seen map[string]struct{} + seenMutex sync.RWMutex + client *statsd.Client } func (s *statsStatsd) IngressTraffic(traffic int) { - s.client.Count("traffic.ingress", traffic) + s.gauge("traffic", int64(traffic), tagTrafficIngress) } func (s *statsStatsd) EgressTraffic(traffic int) { - s.client.Count("traffic.egress", traffic) + s.gauge("traffic", int64(traffic), tagTrafficEgress) } func (s *statsStatsd) ClientConnected(connectionType conntypes.ConnectionType, addr *net.TCPAddr) { @@ -32,25 +83,25 @@ func (s *statsStatsd) ClientDisconnected(connectionType conntypes.ConnectionType s.changeConnections(connectionType, addr, -1) } -func (s *statsStatsd) changeConnections(connectionType conntypes.ConnectionType, addr *net.TCPAddr, value int) { - labels := [...]string{ - "connections", - "intermediate", - "ipv4", - } +func (s *statsStatsd) changeConnections(connectionType conntypes.ConnectionType, addr *net.TCPAddr, increment int64) { + tags := make([]*statsStatsdTag, 0, 2) switch connectionType { case conntypes.ConnectionTypeAbridged: - labels[1] = "abridged" - case conntypes.ConnectionTypeSecure: - labels[1] = "secured" + tags = append(tags, tagConnectionTypeAbridged) + case conntypes.ConnectionTypeIntermediate: + tags = append(tags, tagConnectionTypeIntermediate) + default: + tags = append(tags, tagConnectionTypeSecured) } if addr.IP.To4() == nil { - labels[2] = "ipv6" + tags = append(tags, tagConnectionProtocol6) + } else { + tags = append(tags, tagConnectionProtocol4) } - s.client.Count(strings.Join(labels[:], "."), value) + s.gauge("connections", increment, tags...) } func (s *statsStatsd) TelegramConnected(dc conntypes.DC, addr *net.TCPAddr) { @@ -61,51 +112,83 @@ func (s *statsStatsd) TelegramDisconnected(dc conntypes.DC, addr *net.TCPAddr) { s.changeTelegramConnections(dc, addr, -1) } -func (s *statsStatsd) changeTelegramConnections(dc conntypes.DC, addr *net.TCPAddr, value int) { - labels := [...]string{ - "telegram_connections", - strconv.Itoa(int(dc)), - "ipv4", +func (s *statsStatsd) changeTelegramConnections(dc conntypes.DC, addr *net.TCPAddr, increment int64) { + tags := []*statsStatsdTag{ + { + name: "dc" + strconv.Itoa(int(dc)), + tag: statsd.IntTag("dc", int(dc)), + }, } if addr.IP.To4() == nil { - labels[2] = "ipv6" + tags = append(tags, tagConnectionProtocol6) + } else { + tags = append(tags, tagConnectionProtocol4) } - s.client.Count(strings.Join(labels[:], "."), value) + s.gauge("telegram_connections", increment, tags...) } func (s *statsStatsd) Crash() { - s.client.Increment("crashes") + s.gauge("crashes", 1) } func (s *statsStatsd) ReplayDetected() { - s.client.Increment("replay_attacks") + s.gauge("replay_attacks", 1) } -func newStatsStatsd() (Interface, error) { - options := []statsd.Option{ - statsd.Prefix(config.C.StatsNamespace), - statsd.Network(config.C.StatsdNetwork), - statsd.Address(config.C.StatsdAddr.String()), - statsd.TagsFormat(config.C.StatsdTagsFormat), +func (s *statsStatsd) gauge(metric string, value int64, tags ...*statsStatsdTag) { + key, tagList := s.prepareVals(metric, tags) + s.initGauge(metric, key, tagList) + s.client.GaugeDelta(metric, value, tagList...) +} + +func (s *statsStatsd) prepareVals(metric string, tags []*statsStatsdTag) (string, []statsd.Tag) { + tagList := make([]statsd.Tag, len(tags)) + builder := strings.Builder{} + builder.WriteString(metric) + + for i, v := range tags { + builder.WriteRune('.') + builder.WriteString(v.name) + tagList[i] = v.tag } - if len(config.C.StatsdTags) > 0 { - tags := make([]string, len(config.C.StatsdTags)*2) - for k, v := range config.C.StatsdTags { - tags = append(tags, k, v) - } + return builder.String(), tagList +} - options = append(options, statsd.Tags(tags...)) +func (s *statsStatsd) initGauge(metric, key string, tags []statsd.Tag) { + s.seenMutex.RLock() + _, ok := s.seen[key] + s.seenMutex.RUnlock() + + if ok { + return } - client, err := statsd.New(options...) - if err != nil { - return nil, fmt.Errorf("cannot initialize a client: %w", err) + s.seenMutex.Lock() + defer s.seenMutex.Unlock() + + if _, ok = s.seen[key]; !ok { + s.seen[key] = struct{}{} + s.client.Gauge(metric, 0, tags...) + } +} + +func newStatsStatsd() Interface { + prefix := strings.TrimSuffix(config.C.StatsNamespace, ".") + "." + logger := &statsStatsdLogger{ + log: zap.S().Named("stats").Named("statsd"), } return &statsStatsd{ - client: client, - }, nil + seen: make(map[string]struct{}), + client: statsd.NewClient(config.C.StatsdAddr.String(), + statsd.SendLoopCount(2), + statsd.ReconnectInterval(10*time.Second), + statsd.Logger(logger), + statsd.MetricPrefix(prefix), + statsd.TagStyle(config.C.StatsdTagsFormat), + ), + } }