Add support of EventDomainFronting event

events/event_stream.go

 	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 {
 			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:

events/event_stream_test.go

 	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(),

events/init.go

 	EventStart(mtglib.EventStart)
 	EventFinish(mtglib.EventFinish)
 	EventConnectedToDC(mtglib.EventConnectedToDC)
+	EventDomainFronting(mtglib.EventDomainFronting)
 	EventTraffic(mtglib.EventTraffic)
 	EventConcurrencyLimited(mtglib.EventConcurrencyLimited)
 	EventIPBlocklisted(mtglib.EventIPBlocklisted)

events/init_test.go

 	o.Called(evt)
 }
 
+func (o *ObserverMock) EventDomainFronting(evt mtglib.EventDomainFronting) {
+	o.Called(evt)
+}
+
 func (o *ObserverMock) EventTraffic(evt mtglib.EventTraffic) {
 	o.Called(evt)
 }

events/multi_observer.go

 	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))

events/noop.go

 
 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) {}

events/noop_test.go

 			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",
 				observer.EventStart(typedEvt)
 			case mtglib.EventConnectedToDC:
 				observer.EventConnectedToDC(typedEvt)
+			case mtglib.EventDomainFronting:
+				observer.EventDomainFronting(typedEvt)
 			case mtglib.EventFinish:
 				observer.EventFinish(typedEvt)
 			case mtglib.EventConcurrencyLimited:

mtglib/conns.go

 	"time"
 )
 
-type connTelegramTraffic struct {
+type connTraffic struct {
 	net.Conn
 
 	connID string
 	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 {
 	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 {

mtglib/events.go

 	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
 }

mtglib/events_test.go

 	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(),

mtglib/proxy.go

 	}
 
 	ctx.telegramConn = obfuscated2.Conn{
-		Conn: connTelegramTraffic{
+		Conn: connTraffic{
 			Conn:   conn,
 			connID: ctx.connID,
 			stream: p.eventStream,
 	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)
 		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)

stats/pools.go

 
 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)
 }

stats/prometheus.go

 )
 
 type prometheusProcessor struct {
-	streams map[string]streamInfo
+	streams map[string]*streamInfo
 	factory *PrometheusFactory
 }
 
 	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()
 }
 
 		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()
 }
 
 		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) {
 	}()
 
 	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()
 	}
 }
 }
 
 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 {
 
 func (p *PrometheusFactory) Make() events.Observer {
 	return prometheusProcessor{
-		streams: make(map[string]streamInfo),
+		streams: make(map[string]*streamInfo),
 		factory: p,
 	}
 }
 		}, []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{

stats/prometheus_test.go

 	suite.httpListener.Close()
 }
 
-func (suite *PrometheusTestSuite) TestEventStartFinish() {
+func (suite *PrometheusTestSuite) TestTelegramPath() {
 	suite.prometheus.EventStart(mtglib.EventStart{
 		CreatedAt: time.Now(),
 		ConnID:    "connID",
 	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(),

stats/statsd.go

 )
 
 type statsdProcessor struct {
-	streams map[string]streamInfo
+	streams map[string]*streamInfo
 	client  *statsd.Client
 }
 
 	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
 		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,
 		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) {
 		-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),
 func (s StatsdFactory) Make() events.Observer {
 	return statsdProcessor{
 		client:  s.client,
-		streams: make(map[string]streamInfo),
+		streams: make(map[string]*streamInfo),
 	}
 }
 

stats/statsd_test.go

 	"github.com/stretchr/testify/suite"
 )
 
-const statsdSleepTime = 3 * statsd.DefaultFlushInterval
+const statsdSleepTime = 4 * statsd.DefaultFlushInterval
 
 type statsdFakeServer struct {
 	conn  *net.UDPConn
 	suite.statsdServer.Close()
 }
 
-func (suite *StatsdTestSuite) TestEventStartFinish() {
+func (suite *StatsdTestSuite) TestTelegramPath() {
 	suite.statsd.EventStart(mtglib.EventStart{
 		CreatedAt: time.Now(),
 		ConnID:    "connID",
 		"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() {

stats/stream_info.go

 
 import statsd "github.com/smira/go-statsd"
 
-type streamInfo map[string]string
+type streamInfo struct {
+	isDomainFronted bool
+	tags            map[string]string
+}
 
 func (s streamInfo) T(key string) statsd.Tag {
-	return statsd.StringTag(key, s[key])
+	return statsd.StringTag(key, s.tags[key])
 }
 
-func (s streamInfo) Reset() {
-	for k := range s {
-		delete(s, k)
+func (s *streamInfo) Reset() {
+	s.isDomainFronted = false
+
+	for k := range s.tags {
+		delete(s.tags, k)
 	}
 }