From bef14bd0094bba9852c8d1db1febd06134f59cf8 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Mon, 29 Mar 2021 11:56:05 +0300 Subject: [PATCH] Add support of EventDomainFronting event --- events/event_stream.go | 12 ++++--- events/event_stream_test.go | 23 +++++++++++++ events/init.go | 1 + events/init_test.go | 4 +++ events/multi_observer.go | 15 +++++++++ events/noop.go | 1 + events/noop_test.go | 6 ++++ mtglib/conns.go | 6 ++-- mtglib/events.go | 13 ++++++++ mtglib/events_test.go | 10 ++++++ mtglib/proxy.go | 16 ++++++++-- stats/pools.go | 10 +++--- stats/prometheus.go | 64 +++++++++++++++++++++++++++---------- stats/prometheus_test.go | 61 ++++++++++++++++++++++++++++++++++- stats/statsd.go | 50 ++++++++++++++++++++++------- stats/statsd_test.go | 59 ++++++++++++++++++++++++++++++++-- stats/stream_info.go | 19 +++++++---- 17 files changed, 317 insertions(+), 53 deletions(-) diff --git a/events/event_stream.go b/events/event_stream.go index 1726c81..ef71959 100644 --- a/events/event_stream.go +++ b/events/event_stream.go @@ -60,7 +60,7 @@ func NewEventStream(observerFactories []ObserverFactory) mtglib.EventStream { return rv } -func eventStreamProcessor(ctx context.Context, eventChan <-chan mtglib.Event, observer Observer) { +func eventStreamProcessor(ctx context.Context, eventChan <-chan mtglib.Event, observer Observer) { // nolint: cyclop defer observer.Shutdown() for { @@ -69,14 +69,16 @@ func eventStreamProcessor(ctx context.Context, eventChan <-chan mtglib.Event, ob return case evt := <-eventChan: switch typedEvt := evt.(type) { - case mtglib.EventStart: - observer.EventStart(typedEvt) - case mtglib.EventConnectedToDC: - observer.EventConnectedToDC(typedEvt) case mtglib.EventTraffic: observer.EventTraffic(typedEvt) + case mtglib.EventStart: + observer.EventStart(typedEvt) case mtglib.EventFinish: observer.EventFinish(typedEvt) + case mtglib.EventConnectedToDC: + observer.EventConnectedToDC(typedEvt) + case mtglib.EventDomainFronting: + observer.EventDomainFronting(typedEvt) case mtglib.EventIPBlocklisted: observer.EventIPBlocklisted(typedEvt) case mtglib.EventConcurrencyLimited: diff --git a/events/event_stream_test.go b/events/event_stream_test.go index 90dee03..83e6b77 100644 --- a/events/event_stream_test.go +++ b/events/event_stream_test.go @@ -90,6 +90,29 @@ func (suite *EventStreamTestSuite) TestEventConnectedToDC() { time.Sleep(100 * time.Millisecond) } +func (suite *EventStreamTestSuite) TestEventDomainFronting() { + evt := mtglib.EventDomainFronting{ + CreatedAt: time.Now(), + ConnID: "connID", + } + + for _, v := range []*ObserverMock{suite.observerMock1, suite.observerMock2} { + v. + On("EventDomainFronting", mock.Anything). + Once(). + Run(func(args mock.Arguments) { + caught := args.Get(0).(mtglib.EventDomainFronting) + + suite.Equal(evt.CreatedAt, caught.CreatedAt) + suite.Equal(evt.ConnID, caught.ConnID) + suite.Equal(evt.StreamID(), caught.StreamID()) + }) + } + + suite.stream.Send(suite.ctx, evt) + time.Sleep(100 * time.Millisecond) +} + func (suite *EventStreamTestSuite) TestEventTraffic() { evt := mtglib.EventTraffic{ CreatedAt: time.Now(), diff --git a/events/init.go b/events/init.go index 4720a3b..b6dbd22 100644 --- a/events/init.go +++ b/events/init.go @@ -6,6 +6,7 @@ type Observer interface { EventStart(mtglib.EventStart) EventFinish(mtglib.EventFinish) EventConnectedToDC(mtglib.EventConnectedToDC) + EventDomainFronting(mtglib.EventDomainFronting) EventTraffic(mtglib.EventTraffic) EventConcurrencyLimited(mtglib.EventConcurrencyLimited) EventIPBlocklisted(mtglib.EventIPBlocklisted) diff --git a/events/init_test.go b/events/init_test.go index 58e3635..5fccfad 100644 --- a/events/init_test.go +++ b/events/init_test.go @@ -17,6 +17,10 @@ func (o *ObserverMock) EventConnectedToDC(evt mtglib.EventConnectedToDC) { o.Called(evt) } +func (o *ObserverMock) EventDomainFronting(evt mtglib.EventDomainFronting) { + o.Called(evt) +} + func (o *ObserverMock) EventTraffic(evt mtglib.EventTraffic) { o.Called(evt) } diff --git a/events/multi_observer.go b/events/multi_observer.go index 1dec489..cd74567 100644 --- a/events/multi_observer.go +++ b/events/multi_observer.go @@ -40,6 +40,21 @@ func (m multiObserver) EventConnectedToDC(evt mtglib.EventConnectedToDC) { wg.Wait() } +func (m multiObserver) EventDomainFronting(evt mtglib.EventDomainFronting) { + wg := &sync.WaitGroup{} + wg.Add(len(m.observers)) + + for _, v := range m.observers { + go func(obs Observer) { + defer wg.Done() + + obs.EventDomainFronting(evt) + }(v) + } + + wg.Wait() +} + func (m multiObserver) EventTraffic(evt mtglib.EventTraffic) { wg := &sync.WaitGroup{} wg.Add(len(m.observers)) diff --git a/events/noop.go b/events/noop.go index 8dcbf57..27594d8 100644 --- a/events/noop.go +++ b/events/noop.go @@ -19,6 +19,7 @@ type noopObserver struct{} func (n noopObserver) EventStart(_ mtglib.EventStart) {} func (n noopObserver) EventConnectedToDC(_ mtglib.EventConnectedToDC) {} +func (n noopObserver) EventDomainFronting(_ mtglib.EventDomainFronting) {} func (n noopObserver) EventTraffic(_ mtglib.EventTraffic) {} func (n noopObserver) EventFinish(_ mtglib.EventFinish) {} func (n noopObserver) EventConcurrencyLimited(_ mtglib.EventConcurrencyLimited) {} diff --git a/events/noop_test.go b/events/noop_test.go index 2621e38..9a1d2bd 100644 --- a/events/noop_test.go +++ b/events/noop_test.go @@ -31,6 +31,10 @@ func (suite *NoopTestSuite) SetupSuite() { RemoteIP: net.ParseIP("127.1.0.1"), DC: 2, }, + "domain-fronting": mtglib.EventDomainFronting{ + CreatedAt: time.Now(), + ConnID: "connID", + }, "traffic": mtglib.EventTraffic{ CreatedAt: time.Now(), ConnID: "connID", @@ -76,6 +80,8 @@ func (suite *NoopTestSuite) TestObserver() { observer.EventStart(typedEvt) case mtglib.EventConnectedToDC: observer.EventConnectedToDC(typedEvt) + case mtglib.EventDomainFronting: + observer.EventDomainFronting(typedEvt) case mtglib.EventFinish: observer.EventFinish(typedEvt) case mtglib.EventConcurrencyLimited: diff --git a/mtglib/conns.go b/mtglib/conns.go index 07b7086..e6a64e7 100644 --- a/mtglib/conns.go +++ b/mtglib/conns.go @@ -9,7 +9,7 @@ import ( "time" ) -type connTelegramTraffic struct { +type connTraffic struct { net.Conn connID string @@ -17,7 +17,7 @@ type connTelegramTraffic struct { ctx context.Context } -func (c connTelegramTraffic) Read(b []byte) (int, error) { +func (c connTraffic) Read(b []byte) (int, error) { n, err := c.Conn.Read(b) if n > 0 { @@ -32,7 +32,7 @@ func (c connTelegramTraffic) Read(b []byte) (int, error) { return n, err // nolint: wrapcheck } -func (c connTelegramTraffic) Write(b []byte) (int, error) { +func (c connTraffic) Write(b []byte) (int, error) { n, err := c.Conn.Write(b) if n > 0 { diff --git a/mtglib/events.go b/mtglib/events.go index 3a19b5f..e7a5eae 100644 --- a/mtglib/events.go +++ b/mtglib/events.go @@ -62,6 +62,19 @@ func (e EventFinish) Timestamp() time.Time { return e.CreatedAt } +type EventDomainFronting struct { + CreatedAt time.Time + ConnID string +} + +func (e EventDomainFronting) StreamID() string { + return e.ConnID +} + +func (e EventDomainFronting) Timestamp() time.Time { + return e.CreatedAt +} + type EventConcurrencyLimited struct { CreatedAt time.Time } diff --git a/mtglib/events_test.go b/mtglib/events_test.go index 5597fae..c3a2fb9 100644 --- a/mtglib/events_test.go +++ b/mtglib/events_test.go @@ -58,6 +58,16 @@ func (suite *EventsTestSuite) TestEventTraffic() { suite.WithinDuration(time.Now(), evt.Timestamp(), 10*time.Millisecond) } +func (suite *EventsTestSuite) TestEventDomainFronting() { + evt := mtglib.EventDomainFronting{ + CreatedAt: time.Now(), + ConnID: "CONNID", + } + + suite.Equal("CONNID", evt.StreamID()) + suite.WithinDuration(time.Now(), evt.Timestamp(), 10*time.Millisecond) +} + func (suite *EventsTestSuite) TestEventConcurrencyLimited() { evt := mtglib.EventConcurrencyLimited{ CreatedAt: time.Now(), diff --git a/mtglib/proxy.go b/mtglib/proxy.go index 6baf4a2..731c68c 100644 --- a/mtglib/proxy.go +++ b/mtglib/proxy.go @@ -215,7 +215,7 @@ func (p *Proxy) doTelegramCall(ctx *streamContext) error { } ctx.telegramConn = obfuscated2.Conn{ - Conn: connTelegramTraffic{ + Conn: connTraffic{ Conn: conn, connID: ctx.connID, stream: p.eventStream, @@ -235,7 +235,12 @@ func (p *Proxy) doTelegramCall(ctx *streamContext) error { return nil } -func (p *Proxy) doDomainFronting(ctx context.Context, conn *connRewind) { +func (p *Proxy) doDomainFronting(ctx *streamContext, conn *connRewind) { + p.eventStream.Send(p.ctx, EventDomainFronting{ + CreatedAt: time.Now(), + ConnID: ctx.connID, + }) + conn.Rewind() frontConn, err := p.network.DialContext(ctx, "tcp", p.domainFrontAddress) @@ -245,6 +250,13 @@ func (p *Proxy) doDomainFronting(ctx context.Context, conn *connRewind) { return } + frontConn = connTraffic{ + Conn: frontConn, + ctx: ctx, + connID: ctx.connID, + stream: p.eventStream, + } + rel := relay.AcquireRelay(ctx, p.logger.Named("domain-fronting"), p.bufferSize, p.idleTimeout) defer relay.ReleaseRelay(rel) diff --git a/stats/pools.go b/stats/pools.go index ec0505c..b9b6255 100644 --- a/stats/pools.go +++ b/stats/pools.go @@ -4,15 +4,17 @@ import "sync" var streamInfoPool = sync.Pool{ New: func() interface{} { - return streamInfo{} + return &streamInfo{ + tags: make(map[string]string), + } }, } -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 51c4377..2e7fe10 100644 --- a/stats/prometheus.go +++ b/stats/prometheus.go @@ -13,7 +13,7 @@ import ( ) type prometheusProcessor struct { - streams map[string]streamInfo + streams map[string]*streamInfo factory *PrometheusFactory } @@ -21,15 +21,15 @@ func (p prometheusProcessor) EventStart(evt mtglib.EventStart) { info := acquireStreamInfo() if evt.RemoteIP.To4() != nil { - info[TagIPFamily] = TagIPFamilyIPv4 + info.tags[TagIPFamily] = TagIPFamilyIPv4 } else { - info[TagIPFamily] = TagIPFamilyIPv6 + info.tags[TagIPFamily] = TagIPFamilyIPv6 } p.streams[evt.StreamID()] = info p.factory.metricClientConnections. - WithLabelValues(info[TagIPFamily]). + WithLabelValues(info.tags[TagIPFamily]). Inc() } @@ -39,11 +39,25 @@ func (p prometheusProcessor) EventConnectedToDC(evt mtglib.EventConnectedToDC) { return } - info[TagTelegramIP] = evt.RemoteIP.String() - info[TagDC] = strconv.Itoa(evt.DC) + info.tags[TagTelegramIP] = evt.RemoteIP.String() + info.tags[TagDC] = strconv.Itoa(evt.DC) p.factory.metricTelegramConnections. - WithLabelValues(info[TagTelegramIP], info[TagDC]). + WithLabelValues(info.tags[TagTelegramIP], info.tags[TagDC]). + Inc() +} + +func (p prometheusProcessor) EventDomainFronting(evt mtglib.EventDomainFronting) { + info, ok := p.streams[evt.StreamID()] + if !ok { + return + } + + info.isDomainFronted = true + + p.factory.metricDomainFronting.Inc() + p.factory.metricDomainFrontingConnections. + WithLabelValues(info.tags[TagIPFamily]). Inc() } @@ -53,9 +67,17 @@ func (p prometheusProcessor) EventTraffic(evt mtglib.EventTraffic) { return } - p.factory.metricTelegramTraffic. - WithLabelValues(info[TagTelegramIP], info[TagDC], getDirection(evt.IsRead)). - Add(float64(evt.Traffic)) + direction := getDirection(evt.IsRead) + + if info.isDomainFronted { + p.factory.metricDomainFrontingTraffic. + WithLabelValues(direction). + Add(float64(evt.Traffic)) + } else { + p.factory.metricTelegramTraffic. + WithLabelValues(info.tags[TagTelegramIP], info.tags[TagDC], direction). + Add(float64(evt.Traffic)) + } } func (p prometheusProcessor) EventFinish(evt mtglib.EventFinish) { @@ -70,12 +92,16 @@ func (p prometheusProcessor) EventFinish(evt mtglib.EventFinish) { }() p.factory.metricClientConnections. - WithLabelValues(info[TagIPFamily]). + WithLabelValues(info.tags[TagIPFamily]). Dec() - if telegramIP, ok := info[TagTelegramIP]; ok { + if info.isDomainFronted { + p.factory.metricDomainFrontingConnections. + WithLabelValues(info.tags[TagIPFamily]). + Dec() + } else if telegramIP, ok := info.tags[TagTelegramIP]; ok { p.factory.metricTelegramConnections. - WithLabelValues(telegramIP, info[TagDC]). + WithLabelValues(telegramIP, info.tags[TagDC]). Dec() } } @@ -89,7 +115,11 @@ func (p prometheusProcessor) EventIPBlocklisted(evt mtglib.EventIPBlocklisted) { } func (p prometheusProcessor) Shutdown() { - p.streams = make(map[string]streamInfo) + for _, v := range p.streams { + releaseStreamInfo(v) + } + + p.streams = make(map[string]*streamInfo) } type PrometheusFactory struct { @@ -110,7 +140,7 @@ type PrometheusFactory struct { func (p *PrometheusFactory) Make() events.Observer { return prometheusProcessor{ - streams: make(map[string]streamInfo), + streams: make(map[string]*streamInfo), factory: p, } } @@ -149,8 +179,8 @@ func NewPrometheus(metricPrefix, httpPath string) *PrometheusFactory { // nolint }, []string{TagTelegramIP, TagDC}), metricDomainFrontingConnections: prometheus.NewGaugeVec(prometheus.GaugeOpts{ Namespace: metricPrefix, - Name: MetricDomainFronting, - Help: "A number of connections which talk with front domain.", + Name: MetricDomainFrontingConnections, + Help: "A number of connections which talk to front domain.", }, []string{TagIPFamily}), metricTelegramTraffic: prometheus.NewCounterVec(prometheus.CounterOpts{ diff --git a/stats/prometheus_test.go b/stats/prometheus_test.go index b4a2335..242fe75 100644 --- a/stats/prometheus_test.go +++ b/stats/prometheus_test.go @@ -54,7 +54,7 @@ func (suite *PrometheusTestSuite) TearDownTest() { suite.httpListener.Close() } -func (suite *PrometheusTestSuite) TestEventStartFinish() { +func (suite *PrometheusTestSuite) TestTelegramPath() { suite.prometheus.EventStart(mtglib.EventStart{ CreatedAt: time.Now(), ConnID: "connID", @@ -114,6 +114,65 @@ func (suite *PrometheusTestSuite) TestEventStartFinish() { suite.Contains(data, `mtg_telegram_connections{dc="4",telegram_ip="10.0.0.1"} 0`) } +func (suite *PrometheusTestSuite) TestDomainFrontingPath() { + suite.prometheus.EventStart(mtglib.EventStart{ + CreatedAt: time.Now(), + ConnID: "connID", + RemoteIP: net.ParseIP("10.0.0.10"), + }) + time.Sleep(100 * time.Millisecond) + + data, err := suite.Get() + suite.NoError(err) + suite.Contains(data, `mtg_client_connections{ip_family="ipv4"} 1`) + + suite.prometheus.EventDomainFronting(mtglib.EventDomainFronting{ + CreatedAt: time.Now(), + ConnID: "connID", + }) + time.Sleep(100 * time.Millisecond) + + data, err = suite.Get() + suite.NoError(err) + suite.Contains(data, `mtg_domain_fronting 1`) + suite.Contains(data, `mtg_domain_fronting_connections{ip_family="ipv4"} 1`) + + suite.prometheus.EventTraffic(mtglib.EventTraffic{ + CreatedAt: time.Now(), + ConnID: "connID", + Traffic: 200, + IsRead: true, + }) + time.Sleep(100 * time.Millisecond) + + data, err = suite.Get() + suite.NoError(err) + suite.Contains(data, `mtg_domain_fronting_traffic{direction="to_client"} 200`) + + suite.prometheus.EventTraffic(mtglib.EventTraffic{ + CreatedAt: time.Now(), + ConnID: "connID", + Traffic: 100, + IsRead: false, + }) + time.Sleep(100 * time.Millisecond) + + data, err = suite.Get() + suite.NoError(err) + suite.Contains(data, `mtg_domain_fronting_traffic{direction="from_client"} 100`) + + suite.prometheus.EventFinish(mtglib.EventFinish{ + CreatedAt: time.Now(), + ConnID: "connID", + }) + time.Sleep(100 * time.Millisecond) + + data, err = suite.Get() + suite.NoError(err) + suite.Contains(data, `mtg_client_connections{ip_family="ipv4"} 0`) + suite.Contains(data, `mtg_domain_fronting_connections{ip_family="ipv4"} 0`) +} + func (suite *PrometheusTestSuite) TestEventConcurrencyLimited() { suite.prometheus.EventConcurrencyLimited(mtglib.EventConcurrencyLimited{ CreatedAt: time.Now(), diff --git a/stats/statsd.go b/stats/statsd.go index 44567e3..5bb2a38 100644 --- a/stats/statsd.go +++ b/stats/statsd.go @@ -13,7 +13,7 @@ import ( ) type statsdProcessor struct { - streams map[string]streamInfo + streams map[string]*streamInfo client *statsd.Client } @@ -21,9 +21,9 @@ func (s statsdProcessor) EventStart(evt mtglib.EventStart) { info := acquireStreamInfo() if evt.RemoteIP.To4() != nil { - info[TagIPFamily] = TagIPFamilyIPv4 + info.tags[TagIPFamily] = TagIPFamilyIPv4 } else { - info[TagIPFamily] = TagIPFamilyIPv6 + info.tags[TagIPFamily] = TagIPFamilyIPv6 } s.streams[evt.StreamID()] = info @@ -39,8 +39,8 @@ func (s statsdProcessor) EventConnectedToDC(evt mtglib.EventConnectedToDC) { return } - info[TagTelegramIP] = evt.RemoteIP.String() - info[TagDC] = strconv.Itoa(evt.DC) + info.tags[TagTelegramIP] = evt.RemoteIP.String() + info.tags[TagDC] = strconv.Itoa(evt.DC) s.client.GaugeDelta(MetricTelegramConnections, 1, @@ -48,17 +48,39 @@ func (s statsdProcessor) EventConnectedToDC(evt mtglib.EventConnectedToDC) { info.T(TagDC)) } +func (s statsdProcessor) EventDomainFronting(evt mtglib.EventDomainFronting) { + info, ok := s.streams[evt.StreamID()] + if !ok { + return + } + + info.isDomainFronted = true + + s.client.Incr(MetricDomainFronting, 1) + s.client.GaugeDelta(MetricDomainFrontingConnections, + 1, + info.T(TagIPFamily)) +} + func (s statsdProcessor) EventTraffic(evt mtglib.EventTraffic) { info, ok := s.streams[evt.StreamID()] if !ok { return } - s.client.Incr(MetricTelegramTraffic, - int64(evt.Traffic), - info.T(TagTelegramIP), - info.T(TagDC), - statsd.StringTag(TagDirection, getDirection(evt.IsRead))) + directionTag := statsd.StringTag(TagDirection, getDirection(evt.IsRead)) + + if info.isDomainFronted { + s.client.Incr(MetricDomainFrontingTraffic, + int64(evt.Traffic), + directionTag) + } else { + s.client.Incr(MetricTelegramTraffic, + int64(evt.Traffic), + info.T(TagTelegramIP), + info.T(TagDC), + directionTag) + } } func (s statsdProcessor) EventFinish(evt mtglib.EventFinish) { @@ -76,7 +98,11 @@ func (s statsdProcessor) EventFinish(evt mtglib.EventFinish) { -1, info.T(TagIPFamily)) - if _, ok := info[TagTelegramIP]; ok { + if info.isDomainFronted { + s.client.GaugeDelta(MetricDomainFrontingConnections, + -1, + info.T(TagIPFamily)) + } else if _, ok := info.tags[TagTelegramIP]; ok { s.client.GaugeDelta(MetricTelegramConnections, -1, info.T(TagTelegramIP), @@ -119,7 +145,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/statsd_test.go b/stats/statsd_test.go index 5141eca..ee55ad8 100644 --- a/stats/statsd_test.go +++ b/stats/statsd_test.go @@ -16,7 +16,7 @@ import ( "github.com/stretchr/testify/suite" ) -const statsdSleepTime = 3 * statsd.DefaultFlushInterval +const statsdSleepTime = 4 * statsd.DefaultFlushInterval type statsdFakeServer struct { conn *net.UDPConn @@ -104,7 +104,7 @@ func (suite *StatsdTestSuite) TearDownTest() { suite.statsdServer.Close() } -func (suite *StatsdTestSuite) TestEventStartFinish() { +func (suite *StatsdTestSuite) TestTelegramPath() { suite.statsd.EventStart(mtglib.EventStart{ CreatedAt: time.Now(), ConnID: "connID", @@ -152,6 +152,61 @@ func (suite *StatsdTestSuite) TestEventStartFinish() { "mtg.telegram_connections:-1|g|#telegram_ip:10.1.0.10,dc:2") suite.Contains(suite.statsdServer.String(), "mtg.client_connections:-1|g|#ip_family:ipv4") + + suite.NotContains(suite.statsdServer.String(), "domain_fronting_traffic") + suite.NotContains(suite.statsdServer.String(), "domain_fronting_connections") +} + +func (suite *StatsdTestSuite) TestDomainFrontingPath() { + suite.statsd.EventStart(mtglib.EventStart{ + CreatedAt: time.Now(), + ConnID: "connID", + RemoteIP: net.ParseIP("10.0.0.10"), + }) + time.Sleep(statsdSleepTime) + suite.Equal("mtg.client_connections:+1|g|#ip_family:ipv4", suite.statsdServer.String()) + + suite.statsd.EventDomainFronting(mtglib.EventDomainFronting{ + CreatedAt: time.Now(), + ConnID: "connID", + }) + time.Sleep(statsdSleepTime) + suite.Contains(suite.statsdServer.String(), "mtg.domain_fronting:1|c") + suite.Contains(suite.statsdServer.String(), + `mtg.domain_fronting_connections:+1|g|#ip_family:ipv4`) + + suite.statsd.EventTraffic(mtglib.EventTraffic{ + CreatedAt: time.Now(), + ConnID: "connID", + Traffic: 30, + IsRead: true, + }) + time.Sleep(statsdSleepTime) + suite.Contains(suite.statsdServer.String(), + `mtg.domain_fronting_traffic:30|c|#direction:to_client`) + + suite.statsd.EventTraffic(mtglib.EventTraffic{ + CreatedAt: time.Now(), + ConnID: "connID", + Traffic: 90, + IsRead: false, + }) + time.Sleep(statsdSleepTime) + suite.Contains(suite.statsdServer.String(), + `mtg.domain_fronting_traffic:90|c|#direction:from_client`) + + suite.statsd.EventFinish(mtglib.EventFinish{ + CreatedAt: time.Now(), + ConnID: "connID", + }) + time.Sleep(statsdSleepTime) + suite.Contains(suite.statsdServer.String(), + "mtg.domain_fronting_connections:-1|g|#ip_family:ipv4") + suite.Contains(suite.statsdServer.String(), + "mtg.client_connections:-1|g|#ip_family:ipv4") + + suite.NotContains(suite.statsdServer.String(), "telegram_traffic") + suite.NotContains(suite.statsdServer.String(), "telegram_connections") } func (suite *StatsdTestSuite) TestEventConcurrencyLimited() { diff --git a/stats/stream_info.go b/stats/stream_info.go index 9ada4ff..b28bccb 100644 --- a/stats/stream_info.go +++ b/stats/stream_info.go @@ -2,15 +2,20 @@ package stats import statsd "github.com/smira/go-statsd" -type streamInfo map[string]string - -func (s streamInfo) T(key string) statsd.Tag { - return statsd.StringTag(key, s[key]) +type streamInfo struct { + isDomainFronted bool + tags map[string]string } -func (s streamInfo) Reset() { - for k := range s { - delete(s, k) +func (s streamInfo) T(key string) statsd.Tag { + return statsd.StringTag(key, s.tags[key]) +} + +func (s *streamInfo) Reset() { + s.isDomainFronted = false + + for k := range s.tags { + delete(s.tags, k) } }