ScuroNeko/mtg
1
0
Fork 0
mirror of https://github.com/ScuroNeko/mtg.git synced 2026-08-31 13:54:02 +03:00
Code Issues Packages Projects Releases Wiki Activity

Rename metrics

Browse Source
This commit is contained in:
9seconds
2021-03-29 11:03:54 +03:00
parent 4c3f42e264
commit 36d695118e
5 changed files with 75 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
stats/init.go
+6 -6
View File
@@ -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"
stats/pools.go
+4 -6
View File
@@ -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)
}
stats/prometheus.go
+37 -31
View File
@@ -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)
stats/statsd.go
+21 -15
View File
@@ -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),
}
}
stats/stream_info.go
+7 -42
View File
@@ -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)
}
}