diff --git a/stats/init.go b/stats/init.go index ddeea9d..d9d64b9 100644 --- a/stats/init.go +++ b/stats/init.go @@ -6,14 +6,14 @@ const ( DefaultStatsdMetricPrefix = DefaultMetricPrefix + "." DefaultStatsdTagFormat = "datadog" - MetricClientConnections = "client_connections" - MetricTelegramConnections = "telegram_connections" - MetricDomainDisguisingConnections = "domain_disguising_connections" + MetricClientConnections = "client_connections" + MetricTelegramConnections = "telegram_connections" + MetricDomainFrontingConnections = "domain_fronting_connections" - MetricTelegramTraffic = "telegram_traffic" - MetricDomainDisguisingTraffic = "domain_disguising_traffic" + MetricTelegramTraffic = "telegram_traffic" + MetricDomainFrontingTraffic = "domain_fronting_traffic" - MetricDomainDisguising = "domain_disguising" + MetricDomainFronting = "domain_fronting" MetricConcurrencyLimited = "concurrency_limited" MetricIPBlocklisted = "ip_blocklisted" MetricReplayAttacks = "replay_attacks" diff --git a/stats/pools.go b/stats/pools.go index aaa6c95..ec0505c 100644 --- a/stats/pools.go +++ b/stats/pools.go @@ -4,17 +4,15 @@ import "sync" var streamInfoPool = sync.Pool{ New: func() interface{} { - return &streamInfo{ - tags: map[string]string{}, - } + return streamInfo{} }, } -func acquireStreamInfo() *streamInfo { - return streamInfoPool.Get().(*streamInfo) +func acquireStreamInfo() streamInfo { + return streamInfoPool.Get().(streamInfo) } -func releaseStreamInfo(info *streamInfo) { +func releaseStreamInfo(info streamInfo) { info.Reset() streamInfoPool.Put(info) } diff --git a/stats/prometheus.go b/stats/prometheus.go index 89b12ad..51c4377 100644 --- a/stats/prometheus.go +++ b/stats/prometheus.go @@ -4,6 +4,7 @@ import ( "context" "net" "net/http" + "strconv" "github.com/9seconds/mtg/v2/events" "github.com/9seconds/mtg/v2/mtglib" @@ -12,18 +13,23 @@ import ( ) type prometheusProcessor struct { - streams map[string]*streamInfo + streams map[string]streamInfo factory *PrometheusFactory } func (p prometheusProcessor) EventStart(evt mtglib.EventStart) { info := acquireStreamInfo() - info.SetStartTime(evt.CreatedAt) - info.SetClientIP(evt.RemoteIP) + + if evt.RemoteIP.To4() != nil { + info[TagIPFamily] = TagIPFamilyIPv4 + } else { + info[TagIPFamily] = TagIPFamilyIPv6 + } + p.streams[evt.StreamID()] = info p.factory.metricClientConnections. - WithLabelValues(info.V(TagIPFamily)). + WithLabelValues(info[TagIPFamily]). Inc() } @@ -33,11 +39,11 @@ func (p prometheusProcessor) EventConnectedToDC(evt mtglib.EventConnectedToDC) { return } - info.SetTelegramIP(evt.RemoteIP) - info.SetDC(evt.DC) + info[TagTelegramIP] = evt.RemoteIP.String() + info[TagDC] = strconv.Itoa(evt.DC) p.factory.metricTelegramConnections. - WithLabelValues(info.V(TagTelegramIP), info.V(TagDC)). + WithLabelValues(info[TagTelegramIP], info[TagDC]). Inc() } @@ -48,7 +54,7 @@ func (p prometheusProcessor) EventTraffic(evt mtglib.EventTraffic) { } p.factory.metricTelegramTraffic. - WithLabelValues(info.V(TagTelegramIP), info.V(TagDC), getDirection(evt.IsRead)). + WithLabelValues(info[TagTelegramIP], info[TagDC], getDirection(evt.IsRead)). Add(float64(evt.Traffic)) } @@ -64,12 +70,12 @@ func (p prometheusProcessor) EventFinish(evt mtglib.EventFinish) { }() p.factory.metricClientConnections. - WithLabelValues(info.V(TagIPFamily)). + WithLabelValues(info[TagIPFamily]). Dec() - if info.V(TagTelegramIP) != "" { + if telegramIP, ok := info[TagTelegramIP]; ok { p.factory.metricTelegramConnections. - WithLabelValues(info.V(TagTelegramIP), info.V(TagDC)). + WithLabelValues(telegramIP, info[TagDC]). Dec() } } @@ -83,20 +89,20 @@ func (p prometheusProcessor) EventIPBlocklisted(evt mtglib.EventIPBlocklisted) { } func (p prometheusProcessor) Shutdown() { - p.streams = make(map[string]*streamInfo) + p.streams = make(map[string]streamInfo) } type PrometheusFactory struct { httpServer *http.Server - metricClientConnections *prometheus.GaugeVec - metricTelegramConnections *prometheus.GaugeVec - metricDomainDisguisingConnections *prometheus.GaugeVec + metricClientConnections *prometheus.GaugeVec + metricTelegramConnections *prometheus.GaugeVec + metricDomainFrontingConnections *prometheus.GaugeVec - metricTelegramTraffic *prometheus.CounterVec - metricDomainDisguisingTraffic *prometheus.CounterVec + metricTelegramTraffic *prometheus.CounterVec + metricDomainFrontingTraffic *prometheus.CounterVec - metricDomainDisguising prometheus.Counter + metricDomainFronting prometheus.Counter metricConcurrencyLimited prometheus.Counter metricIPBlocklisted prometheus.Counter metricReplayAttacks prometheus.Counter @@ -104,7 +110,7 @@ type PrometheusFactory struct { func (p *PrometheusFactory) Make() events.Observer { return prometheusProcessor{ - streams: make(map[string]*streamInfo), + streams: make(map[string]streamInfo), factory: p, } } @@ -141,10 +147,10 @@ func NewPrometheus(metricPrefix, httpPath string) *PrometheusFactory { // nolint Name: MetricTelegramConnections, Help: "A number of connections to Telegram servers.", }, []string{TagTelegramIP, TagDC}), - metricDomainDisguisingConnections: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + metricDomainFrontingConnections: prometheus.NewGaugeVec(prometheus.GaugeOpts{ Namespace: metricPrefix, - Name: MetricDomainDisguisingConnections, - Help: "A number of connections which talk with disguising domain.", + Name: MetricDomainFronting, + Help: "A number of connections which talk with front domain.", }, []string{TagIPFamily}), metricTelegramTraffic: prometheus.NewCounterVec(prometheus.CounterOpts{ @@ -152,16 +158,16 @@ func NewPrometheus(metricPrefix, httpPath string) *PrometheusFactory { // nolint Name: MetricTelegramTraffic, Help: "Traffic which is generated talking with Telegram servers.", }, []string{TagTelegramIP, TagDC, TagDirection}), - metricDomainDisguisingTraffic: prometheus.NewCounterVec(prometheus.CounterOpts{ + metricDomainFrontingTraffic: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: metricPrefix, - Name: MetricDomainDisguisingTraffic, - Help: "Traffic which is generated talking with disguising domain.", + Name: MetricDomainFrontingTraffic, + Help: "Traffic which is generated talking with front domain.", }, []string{TagDirection}), - metricDomainDisguising: prometheus.NewCounter(prometheus.CounterOpts{ + metricDomainFronting: prometheus.NewCounter(prometheus.CounterOpts{ Namespace: metricPrefix, - Name: MetricDomainDisguising, - Help: "A number of routings to disguising domain.", + Name: MetricDomainFronting, + Help: "A number of routings to front domain.", }), metricConcurrencyLimited: prometheus.NewCounter(prometheus.CounterOpts{ Namespace: metricPrefix, @@ -182,12 +188,12 @@ func NewPrometheus(metricPrefix, httpPath string) *PrometheusFactory { // nolint registry.MustRegister(factory.metricClientConnections) registry.MustRegister(factory.metricTelegramConnections) - registry.MustRegister(factory.metricDomainDisguisingConnections) + registry.MustRegister(factory.metricDomainFrontingConnections) registry.MustRegister(factory.metricTelegramTraffic) - registry.MustRegister(factory.metricDomainDisguisingTraffic) + registry.MustRegister(factory.metricDomainFrontingTraffic) - registry.MustRegister(factory.metricDomainDisguising) + registry.MustRegister(factory.metricDomainFronting) registry.MustRegister(factory.metricConcurrencyLimited) registry.MustRegister(factory.metricIPBlocklisted) registry.MustRegister(factory.metricReplayAttacks) diff --git a/stats/statsd.go b/stats/statsd.go index b621c73..44567e3 100644 --- a/stats/statsd.go +++ b/stats/statsd.go @@ -2,6 +2,7 @@ package stats import ( "fmt" + "strconv" "strings" "time" @@ -12,19 +13,24 @@ import ( ) type statsdProcessor struct { - streams map[string]*streamInfo + streams map[string]streamInfo client *statsd.Client } func (s statsdProcessor) EventStart(evt mtglib.EventStart) { info := acquireStreamInfo() - info.SetStartTime(evt.CreatedAt) - info.SetClientIP(evt.RemoteIP) + + if evt.RemoteIP.To4() != nil { + info[TagIPFamily] = TagIPFamilyIPv4 + } else { + info[TagIPFamily] = TagIPFamilyIPv6 + } + s.streams[evt.StreamID()] = info s.client.GaugeDelta(MetricClientConnections, 1, - info.TV(TagIPFamily)) + info.T(TagIPFamily)) } func (s statsdProcessor) EventConnectedToDC(evt mtglib.EventConnectedToDC) { @@ -33,13 +39,13 @@ func (s statsdProcessor) EventConnectedToDC(evt mtglib.EventConnectedToDC) { return } - info.SetTelegramIP(evt.RemoteIP) - info.SetDC(evt.DC) + info[TagTelegramIP] = evt.RemoteIP.String() + info[TagDC] = strconv.Itoa(evt.DC) s.client.GaugeDelta(MetricTelegramConnections, 1, - info.TV(TagTelegramIP), - info.TV(TagDC)) + info.T(TagTelegramIP), + info.T(TagDC)) } func (s statsdProcessor) EventTraffic(evt mtglib.EventTraffic) { @@ -50,8 +56,8 @@ func (s statsdProcessor) EventTraffic(evt mtglib.EventTraffic) { s.client.Incr(MetricTelegramTraffic, int64(evt.Traffic), - info.TV(TagTelegramIP), - info.TV(TagDC), + info.T(TagTelegramIP), + info.T(TagDC), statsd.StringTag(TagDirection, getDirection(evt.IsRead))) } @@ -68,13 +74,13 @@ func (s statsdProcessor) EventFinish(evt mtglib.EventFinish) { s.client.GaugeDelta(MetricClientConnections, -1, - info.TV(TagIPFamily)) + info.T(TagIPFamily)) - if info.V(TagTelegramIP) != "" { + if _, ok := info[TagTelegramIP]; ok { s.client.GaugeDelta(MetricTelegramConnections, -1, - info.TV(TagTelegramIP), - info.TV(TagDC)) + info.T(TagTelegramIP), + info.T(TagDC)) } } @@ -113,7 +119,7 @@ func (s StatsdFactory) Close() error { func (s StatsdFactory) Make() events.Observer { return statsdProcessor{ client: s.client, - streams: make(map[string]*streamInfo), + streams: make(map[string]streamInfo), } } diff --git a/stats/stream_info.go b/stats/stream_info.go index 0aeb904..9ada4ff 100644 --- a/stats/stream_info.go +++ b/stats/stream_info.go @@ -1,51 +1,16 @@ package stats -import ( - "net" - "strconv" - "time" +import statsd "github.com/smira/go-statsd" - statsd "github.com/smira/go-statsd" -) +type streamInfo map[string]string -type streamInfo struct { - startTime time.Time - tags map[string]string +func (s streamInfo) T(key string) statsd.Tag { + return statsd.StringTag(key, s[key]) } -func (s *streamInfo) SetStartTime(tme time.Time) { - s.startTime = tme -} - -func (s *streamInfo) SetClientIP(ip net.IP) { - if ip.To4() != nil { - s.tags[TagIPFamily] = TagIPFamilyIPv4 - } else { - s.tags[TagIPFamily] = TagIPFamilyIPv6 - } -} - -func (s *streamInfo) SetTelegramIP(ip net.IP) { - s.tags[TagTelegramIP] = ip.String() -} - -func (s *streamInfo) SetDC(dc int) { - s.tags[TagDC] = strconv.Itoa(dc) -} - -func (s *streamInfo) V(key string) string { - return s.tags[key] -} - -func (s *streamInfo) TV(key string) statsd.Tag { - return statsd.StringTag(key, s.tags[key]) -} - -func (s *streamInfo) Reset() { - s.startTime = time.Time{} - - for k := range s.tags { - delete(s.tags, k) +func (s streamInfo) Reset() { + for k := range s { + delete(s, k) } }