From ff4be53d942aff04b16ec67e1cd5ee7d12292627 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Mon, 9 Jul 2018 09:14:11 +0300 Subject: [PATCH 1/4] Add base stats --- Gopkg.lock | 8 ++++- Gopkg.toml | 4 +++ main.go | 5 +++- stats/server.go | 39 +++++++++++++++++++++++++ stats/stats.go | 77 +++++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 131 insertions(+), 2 deletions(-) create mode 100644 stats/server.go create mode 100644 stats/stats.go diff --git a/Gopkg.lock b/Gopkg.lock index 88ad0e6..cb43617 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -22,6 +22,12 @@ revision = "346938d642f2ec3594ed81d874461961cd0faa76" version = "v1.1.0" +[[projects]] + branch = "master" + name = "github.com/dustin/go-humanize" + packages = ["."] + revision = "02af3965c54e8cacf948b97fef38925c4120652c" + [[projects]] branch = "master" name = "github.com/juju/errors" @@ -80,6 +86,6 @@ [solve-meta] analyzer-name = "dep" analyzer-version = 1 - inputs-digest = "24afdd6b64331aeba47fed75918d04032e13e404612cac107bad1d68a5038b72" + inputs-digest = "312c9fb15085cbe9660443b15a07981990e1f70ec3ddfcce1b7e6cd5902307da" solver-name = "gps-cdcl" solver-version = 1 diff --git a/Gopkg.toml b/Gopkg.toml index ed9feb4..ec43ce8 100644 --- a/Gopkg.toml +++ b/Gopkg.toml @@ -44,3 +44,7 @@ [[constraint]] name = "github.com/satori/go.uuid" version = "1.2.0" + +[[constraint]] + branch = "master" + name = "github.com/dustin/go-humanize" diff --git a/main.go b/main.go index c0ffba1..2e06549 100644 --- a/main.go +++ b/main.go @@ -16,6 +16,7 @@ import ( "github.com/9seconds/mtg/config" "github.com/9seconds/mtg/proxy" + "github.com/9seconds/mtg/stats" "github.com/juju/errors" ) @@ -115,13 +116,15 @@ func main() { zap.ReplaceGlobals(logger) defer logger.Sync() + printURLs(conf.GetURLs()) + if conf.UseMiddleProxy() { zap.S().Infow("Use middle proxy connection to Telegram") } else { zap.S().Infow("Use direct connection to Telegram") } - printURLs(conf.GetURLs()) + go stats.Start(conf) server := proxy.NewProxy(conf) if err := server.Serve(); err != nil { diff --git a/stats/server.go b/stats/server.go new file mode 100644 index 0000000..c3935d2 --- /dev/null +++ b/stats/server.go @@ -0,0 +1,39 @@ +package stats + +import ( + "encoding/json" + "net/http" + "sync" + "time" + + "github.com/9seconds/mtg/config" +) + +var instance *stats + +func Start(conf *config.Config) { + instance = &stats{ + URLs: conf.GetURLs(), + Uptime: uptime(time.Now()), + speedCurrent: &speed{}, + mutex: &sync.RWMutex{}, + } + + http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + + instance.mutex.Lock() + first, _ := json.Marshal(instance) + instance.mutex.Unlock() + + interm := map[string]interface{}{} + json.Unmarshal(first, &interm) + + encoder := json.NewEncoder(w) + encoder.SetEscapeHTML(false) + encoder.SetIndent("", " ") + encoder.Encode(interm) + }) + + http.ListenAndServe(conf.StatAddr(), nil) +} diff --git a/stats/stats.go b/stats/stats.go new file mode 100644 index 0000000..4754674 --- /dev/null +++ b/stats/stats.go @@ -0,0 +1,77 @@ +package stats + +import ( + "encoding/json" + "fmt" + "strconv" + "sync" + "time" + + humanize "github.com/dustin/go-humanize" + + "github.com/9seconds/mtg/config" +) + +type uptime time.Time + +func (u uptime) MarshalJSON() ([]byte, error) { + duration := time.Since(time.Time(u)) + value := map[string]string{ + "seconds": strconv.Itoa(int(duration.Seconds())), + "human": humanize.Time(time.Time(u)), + } + + return json.Marshal(value) +} + +type trafficValue uint64 + +func (t trafficValue) MarshalJSON() ([]byte, error) { + tv := uint64(t) + value := map[string]interface{}{ + "bytes": tv, + "human": humanize.Bytes(tv), + } + + return json.Marshal(value) +} + +type trafficSpeedValue uint64 + +func (t trafficSpeedValue) MarshalJSON() ([]byte, error) { + speed := uint64(t) + value := map[string]interface{}{ + "bytes/s": speed, + "human": fmt.Sprintf("%s/S", humanize.Bytes(speed)), + } + + return json.Marshal(value) +} + +type connections struct { + All uint32 `json:"all"` + Abridged uint32 `json:"abridged"` + Intermediate uint32 `json:"intermediate"` +} + +type traffic struct { + Ingress trafficValue `json:"ingress"` + Egress trafficValue `json:"egress"` +} + +type speed struct { + Ingress trafficSpeedValue `json:"ingress"` + Egress trafficSpeedValue `json:"egress"` +} + +type stats struct { + URLs config.IPURLs `json:"urls"` + ActiveConnections connections `json:"active_connections"` + AllConnections connections `json:"all_connections"` + Traffic traffic `json:"traffic"` + Speed speed `json:"speed"` + Uptime uptime `json:"uptime"` + + speedCurrent *speed + mutex *sync.RWMutex +} From c66e30042550e06ab43e87088da16abcff9a1f85 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Mon, 9 Jul 2018 11:29:47 +0300 Subject: [PATCH 2/4] Stats management utilities --- proxy/proxy.go | 2 +- stats/channels.go | 147 ++++++++++++++++++++++++++++++++++++++++++++++ stats/server.go | 11 ++-- stats/stats.go | 31 ++++++++-- 4 files changed, 182 insertions(+), 9 deletions(-) create mode 100644 stats/channels.go diff --git a/proxy/proxy.go b/proxy/proxy.go index a8399eb..e0f3a4e 100644 --- a/proxy/proxy.go +++ b/proxy/proxy.go @@ -39,7 +39,7 @@ func (p *Proxy) Serve() error { func (p *Proxy) accept(conn net.Conn) { connID := uuid.NewV4().String() - log := zap.S().With("connection_id", connID) + log := zap.S().With("connection_id", connID).Named("main") defer func() { conn.Close() diff --git a/stats/channels.go b/stats/channels.go new file mode 100644 index 0000000..775bd2e --- /dev/null +++ b/stats/channels.go @@ -0,0 +1,147 @@ +package stats + +import ( + "net" + "sync/atomic" + "time" + + "github.com/9seconds/mtg/mtproto" +) + +const ( + crashesChanLength = 1 + connectionsChanLength = 20 + trafficChanLength = 5000 +) + +var ( + CrashesChan = make(chan struct{}, crashesChanLength) + ConnectionsChan = make(chan *connectionData, connectionsChanLength) + TrafficChan = make(chan *trafficData, trafficChanLength) +) + +type connectionData struct { + connectionType mtproto.ConnectionType + addr *net.TCPAddr + connected bool +} + +type trafficData struct { + traffic int + ingress bool +} + +func crashManager() { + for range CrashesChan { + instance.mutex.RLock() + + instance.Crashes++ + + instance.mutex.RUnlock() + } +} + +func connectionManager() { + for event := range ConnectionsChan { + instance.mutex.RLock() + + isIPv4 := event.addr.IP.To4() == nil + var inc uint32 = 1 + if !event.connected { + inc = ^uint32(0) + } + + switch event.connectionType { + case mtproto.ConnectionTypeAbridged: + if isIPv4 { + atomic.AddUint32(&instance.ActiveConnections.Abridged.IPv4, inc) + if event.connected { + atomic.AddUint32(&instance.AllConnections.Abridged.IPv4, inc) + } + } else { + atomic.AddUint32(&instance.ActiveConnections.Abridged.IPv6, inc) + if event.connected { + atomic.AddUint32(&instance.AllConnections.Abridged.IPv6, inc) + } + } + default: + if isIPv4 { + atomic.AddUint32(&instance.ActiveConnections.Intermediate.IPv4, inc) + if event.connected { + atomic.AddUint32(&instance.AllConnections.Intermediate.IPv4, inc) + } + } else { + atomic.AddUint32(&instance.ActiveConnections.Intermediate.IPv6, inc) + if event.connected { + atomic.AddUint32(&instance.AllConnections.Intermediate.IPv6, inc) + } + } + } + + instance.mutex.RUnlock() + } +} + +func trafficManager() { + speedChan := time.Tick(time.Second) + + for { + select { + case event := <-TrafficChan: + instance.mutex.RLock() + + if event.ingress { + instance.Traffic.Ingress += trafficValue(event.traffic) + instance.speedCurrent.Ingress += trafficSpeedValue(event.traffic) + } else { + instance.Traffic.Egress += trafficValue(event.traffic) + instance.speedCurrent.Egress += trafficSpeedValue(event.traffic) + } + + instance.mutex.RUnlock() + case <-speedChan: + instance.mutex.RLock() + + instance.Speed.Ingress = instance.speedCurrent.Ingress + instance.Speed.Egress = instance.speedCurrent.Egress + instance.speedCurrent.Ingress = trafficSpeedValue(0) + instance.speedCurrent.Egress = trafficSpeedValue(0) + + instance.mutex.RUnlock() + } + } +} + +func NewCrash() { + CrashesChan <- struct{}{} +} + +func ClientConnected(connectionType mtproto.ConnectionType, addr *net.TCPAddr) { + ConnectionsChan <- &connectionData{ + connectionType: connectionType, + addr: addr, + connected: true, + } +} + +func ClientDisconnected(connectionType mtproto.ConnectionType, addr *net.TCPAddr) { + ConnectionsChan <- &connectionData{ + connectionType: connectionType, + addr: addr, + connected: false, + } +} + +func IngressTraffic(traffic int) { + TrafficChan <- &trafficData{ + traffic: traffic, + ingress: true, + } +} + +func EgressTraffic(traffic int) { + TrafficChan <- &trafficData{ + traffic: traffic, + ingress: false, + } +} diff --git a/stats/server.go b/stats/server.go index c3935d2..3fb6f99 100644 --- a/stats/server.go +++ b/stats/server.go @@ -13,12 +13,15 @@ var instance *stats func Start(conf *config.Config) { instance = &stats{ - URLs: conf.GetURLs(), - Uptime: uptime(time.Now()), - speedCurrent: &speed{}, - mutex: &sync.RWMutex{}, + URLs: conf.GetURLs(), + Uptime: uptime(time.Now()), + mutex: &sync.RWMutex{}, } + go crashManager() + go connectionManager() + go trafficManager() + http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") diff --git a/stats/stats.go b/stats/stats.go index 4754674..5c6099c 100644 --- a/stats/stats.go +++ b/stats/stats.go @@ -49,9 +49,31 @@ func (t trafficSpeedValue) MarshalJSON() ([]byte, error) { } type connections struct { - All uint32 `json:"all"` - Abridged uint32 `json:"abridged"` - Intermediate uint32 `json:"intermediate"` + All connectionType `json:"all"` + Abridged connectionType `json:"abridged"` + Intermediate connectionType `json:"intermediate"` +} + +func (c connections) MarshalJSON() ([]byte, error) { + c.All.IPv4 = c.Abridged.IPv4 + c.Intermediate.IPv4 + c.All.IPv6 = c.Abridged.IPv6 + c.Intermediate.IPv6 + + value := struct { + All connectionType `json:"all"` + Abridged connectionType `json:"abridged"` + Intermediate connectionType `json:"intermediate"` + }{ + All: c.All, + Abridged: c.Abridged, + Intermediate: c.Intermediate, + } + + return json.Marshal(value) +} + +type connectionType struct { + IPv6 uint32 `json:"ipv6"` + IPv4 uint32 `json:"ipv4"` } type traffic struct { @@ -71,7 +93,8 @@ type stats struct { Traffic traffic `json:"traffic"` Speed speed `json:"speed"` Uptime uptime `json:"uptime"` + Crashes uint32 `json:"crashes"` - speedCurrent *speed + speedCurrent speed mutex *sync.RWMutex } From 045d417ecece26d972d3b14c4949bd95b39f3adb Mon Sep 17 00:00:00 2001 From: 9seconds Date: Mon, 9 Jul 2018 11:32:42 +0300 Subject: [PATCH 3/4] Add logging to stats --- stats/server.go | 20 +++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/stats/server.go b/stats/server.go index 3fb6f99..d0a13b9 100644 --- a/stats/server.go +++ b/stats/server.go @@ -6,12 +6,16 @@ import ( "sync" "time" + "go.uber.org/zap" + "github.com/9seconds/mtg/config" ) var instance *stats func Start(conf *config.Config) { + log := zap.S().Named("stats") + instance = &stats{ URLs: conf.GetURLs(), Uptime: uptime(time.Now()), @@ -26,17 +30,27 @@ func Start(conf *config.Config) { w.Header().Set("Content-Type", "application/json") instance.mutex.Lock() - first, _ := json.Marshal(instance) + first, err := json.Marshal(instance) instance.mutex.Unlock() + if err != nil { + log.Errorw("Cannot encode json", "error", err) + http.Error(w, "Internal server error", 500) + return + } + interm := map[string]interface{}{} json.Unmarshal(first, &interm) encoder := json.NewEncoder(w) encoder.SetEscapeHTML(false) encoder.SetIndent("", " ") - encoder.Encode(interm) + if err = encoder.Encode(interm); err != nil { + log.Errorw("Cannot encode json", "error", err) + } }) - http.ListenAndServe(conf.StatAddr(), nil) + if err := http.ListenAndServe(conf.StatAddr(), nil); err != nil { + log.Fatalw("Stats server has been stopped", "error", err) + } } From 1aa8cfe036e7e437187ee54dd7d773e4f62ab9a3 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Mon, 9 Jul 2018 11:42:44 +0300 Subject: [PATCH 4/4] Stats callbacks --- proxy/proxy.go | 5 +++++ stats/channels.go | 2 +- wrappers/conn.go | 11 ++++++++--- 3 files changed, 14 insertions(+), 4 deletions(-) diff --git a/proxy/proxy.go b/proxy/proxy.go index e0f3a4e..1f2df20 100644 --- a/proxy/proxy.go +++ b/proxy/proxy.go @@ -12,6 +12,7 @@ import ( "github.com/9seconds/mtg/client" "github.com/9seconds/mtg/config" "github.com/9seconds/mtg/mtproto" + "github.com/9seconds/mtg/stats" "github.com/9seconds/mtg/telegram" "github.com/9seconds/mtg/wrappers" ) @@ -45,6 +46,7 @@ func (p *Proxy) accept(conn net.Conn) { conn.Close() if err := recover(); err != nil { + stats.NewCrash() log.Errorw("Crash of accept handler", "error", err) } }() @@ -58,6 +60,9 @@ func (p *Proxy) accept(conn net.Conn) { } defer client.(io.Closer).Close() + stats.ClientConnected(opts.ConnectionType, client.RemoteAddr()) + defer stats.ClientDisconnected(opts.ConnectionType, client.RemoteAddr()) + server, err := p.getTelegramConn(opts, connID) if err != nil { log.Errorw("Cannot initialize server connection", "error", err) diff --git a/stats/channels.go b/stats/channels.go index 775bd2e..1ece855 100644 --- a/stats/channels.go +++ b/stats/channels.go @@ -45,7 +45,7 @@ func connectionManager() { for event := range ConnectionsChan { instance.mutex.RLock() - isIPv4 := event.addr.IP.To4() == nil + isIPv4 := event.addr.IP.To4() != nil var inc uint32 = 1 if !event.connected { inc = ^uint32(0) diff --git a/wrappers/conn.go b/wrappers/conn.go index ad853af..d770cb4 100644 --- a/wrappers/conn.go +++ b/wrappers/conn.go @@ -5,6 +5,8 @@ import ( "time" "go.uber.org/zap" + + "github.com/9seconds/mtg/stats" ) type ConnPurpose uint8 @@ -31,9 +33,10 @@ const ( ) type Conn struct { - connID string - conn net.Conn - logger *zap.SugaredLogger + connID string + conn net.Conn + logger *zap.SugaredLogger + publicIPv4 net.IP publicIPv6 net.IP } @@ -43,6 +46,7 @@ func (c *Conn) Write(p []byte) (int, error) { n, err := c.conn.Write(p) c.logger.Debugw("Write to stream", "bytes", n, "error", err) + stats.EgressTraffic(n) return n, err } @@ -52,6 +56,7 @@ func (c *Conn) Read(p []byte) (int, error) { n, err := c.conn.Read(p) c.logger.Debugw("Read from stream", "bytes", n, "error", err) + stats.IngressTraffic(n) return n, err }