From cb78169d6a72364da9245a7765878b0c7a610c3d Mon Sep 17 00:00:00 2001 From: 9seconds Date: Wed, 10 Oct 2018 13:49:52 +0300 Subject: [PATCH 01/11] Update dependencies --- go.mod | 5 +++-- go.sum | 5 +++++ 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/go.mod b/go.mod index 4abb84d..876eaf5 100644 --- a/go.mod +++ b/go.mod @@ -5,19 +5,20 @@ require ( github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf // indirect github.com/beevik/ntp v0.2.0 github.com/davecgh/go-spew v1.1.1 // indirect - github.com/dustin/go-humanize v0.0.0-20180713052910-9f541cc9db5d + github.com/dustin/go-humanize v1.0.0 github.com/gofrs/uuid v3.1.0+incompatible github.com/juju/errors v0.0.0-20180806074554-22422dad46e1 github.com/juju/loggo v0.0.0-20180524022052-584905176618 // indirect github.com/juju/testing v0.0.0-20180920084828-472a3e8b2073 // indirect github.com/kr/pretty v0.1.0 // indirect + github.com/kr/pty v1.1.3 // indirect github.com/pkg/errors v0.8.0 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/stretchr/testify v1.2.2 go.uber.org/atomic v1.3.2 // indirect go.uber.org/multierr v1.1.0 // indirect go.uber.org/zap v1.9.1 - golang.org/x/net v0.0.0-20180921000356-2f5d2388922f // indirect + golang.org/x/net v0.0.0-20181005035420-146acd28ed58 // indirect gopkg.in/alecthomas/kingpin.v2 v2.2.6 gopkg.in/alexcesaro/statsd.v2 v2.0.0 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect diff --git a/go.sum b/go.sum index f145ee0..632dd28 100644 --- a/go.sum +++ b/go.sum @@ -8,6 +8,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dustin/go-humanize v0.0.0-20180713052910-9f541cc9db5d h1:lDrio3iIdNb0Gw9CgH7cQF+iuB5mOOjdJ9ERNJCBgb4= github.com/dustin/go-humanize v0.0.0-20180713052910-9f541cc9db5d/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= +github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo= +github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= github.com/gofrs/uuid v3.1.0+incompatible h1:q2rtkjaKT4YEr6E1kamy0Ha4RtepWlQBedyHx0uzKwA= github.com/gofrs/uuid v3.1.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM= github.com/juju/errors v0.0.0-20180806074554-22422dad46e1 h1:wnhMXidtb70kDZCeLt/EfsVtkXS5c8zLnE9y/6DIRAU= @@ -19,6 +21,7 @@ github.com/juju/testing v0.0.0-20180920084828-472a3e8b2073/go.mod h1:63prj8cnj0t github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/pty v1.1.3/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw= @@ -35,6 +38,8 @@ go.uber.org/zap v1.9.1 h1:XCJQEf3W6eZaVwhRBof6ImoYGJSITeKWsyeh3HFu/5o= go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q= golang.org/x/net v0.0.0-20180921000356-2f5d2388922f h1:QM2QVxvDoW9PFSPp/zy9FgxJLfaWTZlS61KEPtBwacM= golang.org/x/net v0.0.0-20180921000356-2f5d2388922f/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/net v0.0.0-20181005035420-146acd28ed58 h1:otZG8yDCO4LVps5+9bxOeNiCvgmOyt96J3roHTYs7oE= +golang.org/x/net v0.0.0-20181005035420-146acd28ed58/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/alexcesaro/statsd.v2 v2.0.0 h1:FXkZSCZIH17vLCO5sO2UucTHsH9pc+17F6pl3JVCwMc= From 76cfbb009a15e9ac57d279d2358dcc97d396ecb1 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Fri, 12 Oct 2018 10:38:35 +0300 Subject: [PATCH 02/11] Decrease an amount of goroutines --- proxy/proxy.go | 22 +++++-------------- wrappers/conn.go | 57 +++++++++++++++++------------------------------- 2 files changed, 26 insertions(+), 53 deletions(-) diff --git a/proxy/proxy.go b/proxy/proxy.go index 96054e4..4a7c43c 100644 --- a/proxy/proxy.go +++ b/proxy/proxy.go @@ -93,12 +93,12 @@ func (p *Proxy) accept(conn net.Conn) { clientPacket := clientConn.(wrappers.PacketReadWriteCloser) serverPacket := serverConn.(wrappers.PacketReadWriteCloser) go p.middlePipe(clientPacket, serverPacket, wait, &opts.ReadHacks) - go p.middlePipe(serverPacket, clientPacket, wait, &opts.WriteHacks) + p.middlePipe(serverPacket, clientPacket, wait, &opts.WriteHacks) } else { clientStream := clientConn.(wrappers.StreamReadWriteCloser) serverStream := serverConn.(wrappers.StreamReadWriteCloser) go p.directPipe(clientStream, serverStream, wait, p.conf.ReadBufferSize) - go p.directPipe(serverStream, clientStream, wait, p.conf.WriteBufferSize) + p.directPipe(serverStream, clientStream, wait, p.conf.WriteBufferSize) } wait.Wait() @@ -121,13 +121,8 @@ func (p *Proxy) getTelegramConn(ctx context.Context, cancel context.CancelFunc, return packetConn, nil } -func (p *Proxy) middlePipe(src wrappers.PacketReadCloser, dst io.WriteCloser, - wait *sync.WaitGroup, hacks *mtproto.Hacks) { - defer func() { - src.Close() // nolint: errcheck, gosec - dst.Close() // nolint: errcheck, gosec - wait.Done() - }() +func (p *Proxy) middlePipe(src wrappers.PacketReadCloser, dst io.Writer, wait *sync.WaitGroup, hacks *mtproto.Hacks) { + defer wait.Done() for { hacks.SimpleAck = false @@ -145,13 +140,8 @@ func (p *Proxy) middlePipe(src wrappers.PacketReadCloser, dst io.WriteCloser, } } -func (p *Proxy) directPipe(src wrappers.StreamReadCloser, dst io.WriteCloser, - wait *sync.WaitGroup, bufferSize int) { - defer func() { - src.Close() // nolint: errcheck, gosec - dst.Close() // nolint: errcheck, gosec - wait.Done() - }() +func (p *Proxy) directPipe(src wrappers.StreamReadCloser, dst io.Writer, wait *sync.WaitGroup, bufferSize int) { + defer wait.Done() buffer := make([]byte, bufferSize) if _, err := io.CopyBuffer(dst, src, buffer); err != nil { diff --git a/wrappers/conn.go b/wrappers/conn.go index 157828e..239af53 100644 --- a/wrappers/conn.go +++ b/wrappers/conn.go @@ -38,13 +38,6 @@ const ( connTimeoutWrite = 2 * time.Minute ) -type ioResult struct { - n int - err error -} - -type ioFunc func([]byte) (int, error) - // Conn is a basic wrapper for net.Conn providing the most low-level // logic and management as possible. type Conn struct { @@ -61,12 +54,20 @@ type Conn struct { func (c *Conn) Write(p []byte) (int, error) { select { case <-c.ctx.Done(): + c.Close() // nolint: gosec return 0, errors.Annotate(c.ctx.Err(), "Cannot write because context was closed") default: - n, err := c.doIO(c.conn.Write, p, connTimeoutWrite) + if err := c.conn.SetWriteDeadline(time.Now().Add(connTimeoutWrite)); err != nil { + c.Close() // nolint: gosec + return 0, errors.Annotate(err, "Cannot set write deadline to the socket") + } + n, err := c.conn.Write(p) c.logger.Debugw("Write to stream", "bytes", n, "error", err) stats.EgressTraffic(n) + if err != nil { + c.Close() // nolint: gosec + } return n, err } @@ -75,48 +76,30 @@ func (c *Conn) Write(p []byte) (int, error) { func (c *Conn) Read(p []byte) (int, error) { select { case <-c.ctx.Done(): + c.Close() // nolint: gosec return 0, errors.Annotate(c.ctx.Err(), "Cannot read because context was closed") default: - n, err := c.doIO(c.conn.Read, p, connTimeoutRead) + if err := c.conn.SetReadDeadline(time.Now().Add(connTimeoutRead)); err != nil { + c.Close() // nolint: gosec + return 0, errors.Annotate(err, "Cannot set read deadline to the socket") + } + n, err := c.conn.Read(p) c.logger.Debugw("Read from stream", "bytes", n, "error", err) stats.IngressTraffic(n) + if err != nil { + c.Close() // nolint: gosec + } return n, err } } -func (c *Conn) doIO(callback ioFunc, p []byte, timeout time.Duration) (int, error) { - resChan := make(chan ioResult, 1) - timer := time.NewTimer(timeout) - - go func() { - n, err := callback(p) - resChan <- ioResult{n: n, err: err} - }() - - select { - case res := <-resChan: - timer.Stop() - if res.err != nil { - c.Close() // nolint: gosec - } - return res.n, res.err - case <-c.ctx.Done(): - timer.Stop() - c.Close() // nolint: gosec - return 0, errors.Annotate(c.ctx.Err(), "Cannot do IO because context is closed") - case <-timer.C: - c.Close() // nolint: gosec - return 0, errors.Annotate(c.ctx.Err(), "Timeout on IO operation") - } -} - // Close closes underlying net.Conn instance. func (c *Conn) Close() error { - defer c.logger.Debugw("Close connection") - + c.logger.Debugw("Close connection") c.cancel() + return c.conn.Close() } From cf9bf56d8eb3117f50b2292f172014287e0bd390 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Fri, 12 Oct 2018 10:39:50 +0300 Subject: [PATCH 03/11] Update dependencies --- go.mod | 5 ++--- go.sum | 13 ++++--------- 2 files changed, 6 insertions(+), 12 deletions(-) diff --git a/go.mod b/go.mod index 876eaf5..18ed04a 100644 --- a/go.mod +++ b/go.mod @@ -7,18 +7,17 @@ require ( github.com/davecgh/go-spew v1.1.1 // indirect github.com/dustin/go-humanize v1.0.0 github.com/gofrs/uuid v3.1.0+incompatible - github.com/juju/errors v0.0.0-20180806074554-22422dad46e1 + github.com/juju/errors v0.0.0-20181012004132-a4583d0a56ea github.com/juju/loggo v0.0.0-20180524022052-584905176618 // indirect github.com/juju/testing v0.0.0-20180920084828-472a3e8b2073 // indirect github.com/kr/pretty v0.1.0 // indirect - github.com/kr/pty v1.1.3 // indirect github.com/pkg/errors v0.8.0 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/stretchr/testify v1.2.2 go.uber.org/atomic v1.3.2 // indirect go.uber.org/multierr v1.1.0 // indirect go.uber.org/zap v1.9.1 - golang.org/x/net v0.0.0-20181005035420-146acd28ed58 // indirect + golang.org/x/net v0.0.0-20181011144130-49bb7cea24b1 // indirect gopkg.in/alecthomas/kingpin.v2 v2.2.6 gopkg.in/alexcesaro/statsd.v2 v2.0.0 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect diff --git a/go.sum b/go.sum index 632dd28..33453b3 100644 --- a/go.sum +++ b/go.sum @@ -6,14 +6,12 @@ github.com/beevik/ntp v0.2.0 h1:sGsd+kAXzT0bfVfzJfce04g+dSRfrs+tbQW8lweuYgw= github.com/beevik/ntp v0.2.0/go.mod h1:hIHWr+l3+/clUnF44zdK+CWW7fO8dR5cIylAQ76NRpg= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/dustin/go-humanize v0.0.0-20180713052910-9f541cc9db5d h1:lDrio3iIdNb0Gw9CgH7cQF+iuB5mOOjdJ9ERNJCBgb4= -github.com/dustin/go-humanize v0.0.0-20180713052910-9f541cc9db5d/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo= github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= github.com/gofrs/uuid v3.1.0+incompatible h1:q2rtkjaKT4YEr6E1kamy0Ha4RtepWlQBedyHx0uzKwA= github.com/gofrs/uuid v3.1.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM= -github.com/juju/errors v0.0.0-20180806074554-22422dad46e1 h1:wnhMXidtb70kDZCeLt/EfsVtkXS5c8zLnE9y/6DIRAU= -github.com/juju/errors v0.0.0-20180806074554-22422dad46e1/go.mod h1:W54LbzXuIE0boCoNJfwqpmkKJ1O4TCTZMetAt6jGk7Q= +github.com/juju/errors v0.0.0-20181012004132-a4583d0a56ea h1:g2k+8WR7cHch4g0tBDhfiEvAp7fXxTNBiD1oC1Oxj3E= +github.com/juju/errors v0.0.0-20181012004132-a4583d0a56ea/go.mod h1:W54LbzXuIE0boCoNJfwqpmkKJ1O4TCTZMetAt6jGk7Q= github.com/juju/loggo v0.0.0-20180524022052-584905176618 h1:MK144iBQF9hTSwBW/9eJm034bVoG30IshVm688T2hi8= github.com/juju/loggo v0.0.0-20180524022052-584905176618/go.mod h1:vgyd7OREkbtVEN/8IXZe5Ooef3LQePvuBm9UWj6ZL8U= github.com/juju/testing v0.0.0-20180920084828-472a3e8b2073 h1:WQM1NildKThwdP7qWrNAFGzp4ijNLw8RlgENkaI4MJs= @@ -21,7 +19,6 @@ github.com/juju/testing v0.0.0-20180920084828-472a3e8b2073/go.mod h1:63prj8cnj0t github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= -github.com/kr/pty v1.1.3/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw= @@ -36,10 +33,8 @@ go.uber.org/multierr v1.1.0 h1:HoEmRHQPVSqub6w2z2d2EOVs2fjyFRGyofhKuyDq0QI= go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0= go.uber.org/zap v1.9.1 h1:XCJQEf3W6eZaVwhRBof6ImoYGJSITeKWsyeh3HFu/5o= go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q= -golang.org/x/net v0.0.0-20180921000356-2f5d2388922f h1:QM2QVxvDoW9PFSPp/zy9FgxJLfaWTZlS61KEPtBwacM= -golang.org/x/net v0.0.0-20180921000356-2f5d2388922f/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= -golang.org/x/net v0.0.0-20181005035420-146acd28ed58 h1:otZG8yDCO4LVps5+9bxOeNiCvgmOyt96J3roHTYs7oE= -golang.org/x/net v0.0.0-20181005035420-146acd28ed58/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/net v0.0.0-20181011144130-49bb7cea24b1 h1:Y/KGZSOdz/2r0WJ9Mkmz6NJBusp0kiNx1Cn82lzJQ6w= +golang.org/x/net v0.0.0-20181011144130-49bb7cea24b1/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/alexcesaro/statsd.v2 v2.0.0 h1:FXkZSCZIH17vLCO5sO2UucTHsH9pc+17F6pl3JVCwMc= From 7128e70e50779ed9ee5709e9508563ccc5080f03 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Fri, 12 Oct 2018 10:50:52 +0300 Subject: [PATCH 04/11] Minor changes in dockerfile --- Dockerfile | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/Dockerfile b/Dockerfile index 6c3e2e2..a86a6f0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -10,8 +10,7 @@ RUN set -x \ curl \ git \ make \ - upx \ - && update-ca-certificates + upx COPY . /go/src/github.com/9seconds/mtg/ @@ -26,7 +25,7 @@ RUN set -x \ FROM scratch -ENTRYPOINT ["/usr/local/bin/mtg"] +ENTRYPOINT ["/mtg"] ENV MTG_IP=0.0.0.0 \ MTG_PORT=3128 \ MTG_STATS_IP=0.0.0.0 \ @@ -34,4 +33,4 @@ ENV MTG_IP=0.0.0.0 \ EXPOSE 3128 3129 COPY --from=0 /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt -COPY --from=0 /go/src/github.com/9seconds/mtg/mtg /usr/local/bin/mtg +COPY --from=0 /go/src/github.com/9seconds/mtg/mtg /mtg From 7c463851b2c05c2bba84640830047e93828df4c2 Mon Sep 17 00:00:00 2001 From: Sergey Arkhipov <9seconds@users.noreply.github.com> Date: Sat, 13 Oct 2018 11:18:36 +0300 Subject: [PATCH 05/11] Okay, 3 megabytes. Image grows :( --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index dd7e636..ecf09f3 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,7 @@ mtg is an implementation in golang which is intended to be: software. I also believe that in case of throwout proxies, this feature is useless luxury. * **Minimum docker image size** - Official image is less than 2.5 megabytes. Literally. + Official image is less than 3 megabytes. Literally. * **No management WebUI** This is an implementation of simple lightweight proxy. I won't do that. From c74e0ee359ff6a4016575bc4e4dcf0cc0a4821c2 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Tue, 16 Oct 2018 22:17:46 +0300 Subject: [PATCH 06/11] Rework stats server --- main.go | 2 +- stats/channels.go | 102 +++++----------------------- stats/init.go | 22 ++++++ stats/server.go | 40 ++--------- stats/stats.go | 169 ++++++++++++++++++++++++++++++++-------------- stats/statsd.go | 12 ++-- 6 files changed, 169 insertions(+), 178 deletions(-) create mode 100644 stats/init.go diff --git a/main.go b/main.go index 03857c8..2d5bdcd 100644 --- a/main.go +++ b/main.go @@ -193,7 +193,7 @@ func main() { // nolint: gocyclo zap.S().Infow("Use direct connection to Telegram") } - if err := stats.Start(conf); err != nil { + if err := stats.Init(conf); err != nil { panic(err) } diff --git a/stats/channels.go b/stats/channels.go index 34616cb..bd9d9fb 100644 --- a/stats/channels.go +++ b/stats/channels.go @@ -2,21 +2,20 @@ package stats import ( "net" - "time" "github.com/9seconds/mtg/mtproto" ) const ( - crashesChanLength = 1 - connectionsChanLength = 20 - trafficChanLength = 5000 + connectionsChanLength = 10 + trafficChanLength = 10 ) var ( - crashesChan = make(chan struct{}, crashesChanLength) - connectionsChan = make(chan *connectionData, connectionsChanLength) - trafficChan = make(chan *trafficData, trafficChanLength) + crashesChan = make(chan struct{}) + statsChan = make(chan chan<- Stats) + connectionsChan = make(chan connectionData, connectionsChanLength) + trafficChan = make(chan trafficData, trafficChanLength) ) type connectionData struct { @@ -30,81 +29,6 @@ type trafficData struct { 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 { - instance.Connections.Abridged.IPv4 += inc - } else { - instance.Connections.Abridged.IPv6 += inc - } - case mtproto.ConnectionTypeSecure: - if isIPv4 { - instance.Connections.Secure.IPv4 += inc - } else { - instance.Connections.Secure.IPv6 += inc - } - default: - if isIPv4 { - instance.Connections.Intermediate.IPv4 += inc - } else { - instance.Connections.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() - } - } -} - // NewCrash indicates new crash. func NewCrash() { crashesChan <- struct{}{} @@ -112,7 +36,7 @@ func NewCrash() { // ClientConnected indicates that new client was connected. func ClientConnected(connectionType mtproto.ConnectionType, addr *net.TCPAddr) { - connectionsChan <- &connectionData{ + connectionsChan <- connectionData{ connectionType: connectionType, addr: addr, connected: true, @@ -121,7 +45,7 @@ func ClientConnected(connectionType mtproto.ConnectionType, addr *net.TCPAddr) { // ClientDisconnected indicates that client was disconnected. func ClientDisconnected(connectionType mtproto.ConnectionType, addr *net.TCPAddr) { - connectionsChan <- &connectionData{ + connectionsChan <- connectionData{ connectionType: connectionType, addr: addr, connected: false, @@ -130,7 +54,7 @@ func ClientDisconnected(connectionType mtproto.ConnectionType, addr *net.TCPAddr // IngressTraffic accounts new ingress traffic. func IngressTraffic(traffic int) { - trafficChan <- &trafficData{ + trafficChan <- trafficData{ traffic: traffic, ingress: true, } @@ -138,8 +62,14 @@ func IngressTraffic(traffic int) { // EgressTraffic accounts new ingress traffic. func EgressTraffic(traffic int) { - trafficChan <- &trafficData{ + trafficChan <- trafficData{ traffic: traffic, ingress: false, } } + +func GetStats() Stats { + rpcChan := make(chan Stats) + statsChan <- rpcChan + return <-rpcChan +} diff --git a/stats/init.go b/stats/init.go new file mode 100644 index 0000000..3fb206a --- /dev/null +++ b/stats/init.go @@ -0,0 +1,22 @@ +package stats + +import ( + "github.com/juju/errors" + + "github.com/9seconds/mtg/config" +) + +func Init(conf *config.Config) error { + if conf.StatsD.Enabled { + client, err := newStatsd(conf) + if err != nil { + return errors.Annotate(err, "Cannot initialize statsd client") + } + go client.run() + } + + go NewStats(conf).start() + go startServer(conf) + + return nil +} diff --git a/stats/server.go b/stats/server.go index 69c4417..df17e83 100644 --- a/stats/server.go +++ b/stats/server.go @@ -3,45 +3,19 @@ package stats import ( "encoding/json" "net/http" - "sync" - "time" "go.uber.org/zap" "github.com/9seconds/mtg/config" ) -var instance *stats - -// Start starts new statistics server. -func Start(conf *config.Config) error { +func startServer(conf *config.Config) { log := zap.S().Named("stats") - instance = &stats{ - URLs: conf.GetURLs(), - Uptime: uptime(time.Now()), - mutex: &sync.RWMutex{}, - } - - if conf.StatsD.Enabled { - client, err := newStatsd(conf) - if err != nil { - return err - } - go client.run() - } - - go crashManager() - go connectionManager() - go trafficManager() - http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") - instance.mutex.Lock() - first, err := json.Marshal(instance) - instance.mutex.Unlock() - + first, err := json.Marshal(GetStats()) if err != nil { log.Errorw("Cannot encode json", "error", err) http.Error(w, "Internal server error", 500) @@ -59,11 +33,7 @@ func Start(conf *config.Config) error { } }) - go func() { - if err := http.ListenAndServe(conf.StatAddr(), nil); err != nil { - log.Fatalw("Stats server has been stopped", "error", err) - } - }() - - return nil + if err := http.ListenAndServe(conf.StatAddr(), nil); err != nil { + log.Fatalw("Stats server has been stopped", "error", err) + } } diff --git a/stats/stats.go b/stats/stats.go index ec9994b..e9c5999 100644 --- a/stats/stats.go +++ b/stats/stats.go @@ -4,12 +4,12 @@ import ( "encoding/json" "fmt" "strconv" - "sync" "time" humanize "github.com/dustin/go-humanize" "github.com/9seconds/mtg/config" + "github.com/9seconds/mtg/mtproto" ) type uptime time.Time @@ -24,72 +24,72 @@ func (u uptime) MarshalJSON() ([]byte, error) { 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 connectionType struct { + IPv6 uint32 `json:"ipv6"` + IPv4 uint32 `json:"ipv4"` } -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 { +type baseConnections struct { All connectionType `json:"all"` Abridged connectionType `json:"abridged"` Intermediate connectionType `json:"intermediate"` Secure connectionType `json:"secure"` } +type connections struct { + baseConnections +} + func (c connections) MarshalJSON() ([]byte, error) { c.All.IPv4 = c.Abridged.IPv4 + c.Intermediate.IPv4 + c.Secure.IPv4 c.All.IPv6 = c.Abridged.IPv6 + c.Intermediate.IPv6 + c.Secure.IPv6 - value := struct { - All connectionType `json:"all"` - Abridged connectionType `json:"abridged"` - Intermediate connectionType `json:"intermediate"` - Secure connectionType `json:"secure"` - }{ - All: c.All, - Abridged: c.Abridged, - Intermediate: c.Intermediate, - Secure: c.Secure, + return json.Marshal(c.baseConnections) +} + +type traffic struct { + ingress uint64 + egress uint64 +} + +func (t *traffic) dumpValue(value uint64) map[string]interface{} { + return map[string]interface{}{ + "bytes": value, + "human": humanize.Bytes(value), + } +} + +func (t traffic) MarshalJSON() ([]byte, error) { + value := map[string]map[string]interface{}{ + "ingress": t.dumpValue(t.ingress), + "egress": t.dumpValue(t.egress), } return json.Marshal(value) } -type connectionType struct { - IPv6 uint32 `json:"ipv6"` - IPv4 uint32 `json:"ipv4"` -} - -type traffic struct { - Ingress trafficValue `json:"ingress"` - Egress trafficValue `json:"egress"` -} - type speed struct { - Ingress trafficSpeedValue `json:"ingress"` - Egress trafficSpeedValue `json:"egress"` + ingress uint64 + egress uint64 } -type stats struct { +func (s *speed) dumpValue(value uint64) map[string]interface{} { + return map[string]interface{}{ + "bytes/s": value, + "human": fmt.Sprintf("%s/s", humanize.Bytes(value)), + } +} + +func (s speed) MarshalJSON() ([]byte, error) { + value := map[string]map[string]interface{}{ + "ingress": s.dumpValue(s.ingress), + "egress": s.dumpValue(s.egress), + } + + return json.Marshal(value) +} + +type Stats struct { URLs config.IPURLs `json:"urls"` Connections connections `json:"connections"` Traffic traffic `json:"traffic"` @@ -97,6 +97,77 @@ type stats struct { Uptime uptime `json:"uptime"` Crashes uint32 `json:"crashes"` - speedCurrent speed - mutex *sync.RWMutex + previousTraffic traffic +} + +func (s *Stats) start() { + speedChan := time.Tick(time.Second) + + for { + select { + case <-speedChan: + s.handleSpeed() + case event := <-trafficChan: + s.handleTraffic(event) + case event := <-connectionsChan: + s.handleConnection(event) + case getStatsChan := <-statsChan: + s.handleGetStats(getStatsChan) + case <-crashesChan: + s.handleCrash() + } + } +} + +func (s *Stats) handleTraffic(evt trafficData) { + if evt.ingress { + s.Traffic.ingress += uint64(evt.traffic) + } else { + s.Traffic.egress += uint64(evt.traffic) + } +} + +func (s *Stats) handleSpeed() { + s.Speed.ingress = s.Traffic.ingress - s.previousTraffic.ingress + s.Speed.egress = s.Traffic.egress - s.previousTraffic.egress + s.previousTraffic.ingress = s.Traffic.ingress + s.previousTraffic.egress = s.Traffic.egress +} + +func (s *Stats) handleConnection(evt connectionData) { + var inc uint32 = 1 + if !evt.connected { + inc = ^uint32(0) + } + + var conn *connectionType + switch evt.connectionType { + case mtproto.ConnectionTypeAbridged: + conn = &s.Connections.Abridged + case mtproto.ConnectionTypeSecure: + conn = &s.Connections.Secure + default: + conn = &s.Connections.Intermediate + } + + if evt.addr.IP.To4() != nil { + conn.IPv4 += inc + } else { + conn.IPv6 += inc + } +} + +func (s *Stats) handleGetStats(getStatsChan chan<- Stats) { + getStatsChan <- *s +} + +func (s *Stats) handleCrash() { + s.Crashes++ +} + +func NewStats(conf *config.Config) *Stats { + return &Stats{ + URLs: conf.GetURLs(), + Uptime: uptime(time.Now()), + } } diff --git a/stats/statsd.go b/stats/statsd.go index 2ecaa63..5745889 100644 --- a/stats/statsd.go +++ b/stats/statsd.go @@ -36,7 +36,7 @@ type statsdExporter struct { func (s *statsdExporter) run() { for range time.Tick(statsdPollTime) { - instance.mutex.Lock() + instance := GetStats() s.client.Gauge(statsdConnectionsAbridgedV4, instance.Connections.Abridged.IPv4) s.client.Gauge(statsdConnectionsAbridgedV6, instance.Connections.Abridged.IPv6) @@ -44,13 +44,11 @@ func (s *statsdExporter) run() { s.client.Gauge(statsdConnectionsIntermediateV6, instance.Connections.Intermediate.IPv6) s.client.Gauge(statsdConnectionsSecureV4, instance.Connections.Secure.IPv4) s.client.Gauge(statsdConnectionsSecureV6, instance.Connections.Secure.IPv6) - s.client.Gauge(statsdTrafficIngress, uint64(instance.Traffic.Ingress)) - s.client.Gauge(statsdTrafficEgress, uint64(instance.Traffic.Egress)) - s.client.Gauge(statsdSpeedIngress, uint64(instance.Speed.Ingress)) - s.client.Gauge(statsdSpeedEgress, uint64(instance.Speed.Egress)) + s.client.Gauge(statsdTrafficIngress, instance.Traffic.ingress) + s.client.Gauge(statsdTrafficEgress, instance.Traffic.egress) + s.client.Gauge(statsdSpeedIngress, instance.Speed.ingress) + s.client.Gauge(statsdSpeedEgress, instance.Speed.egress) s.client.Gauge(statsdCrashes, instance.Crashes) - - instance.mutex.Unlock() } } From 358d42ec61f055071e7ba8e369dd92883a4ee0f3 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Wed, 17 Oct 2018 10:49:27 +0300 Subject: [PATCH 07/11] Fix spelling issue --- stats/server.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/stats/server.go b/stats/server.go index df17e83..b4a0fef 100644 --- a/stats/server.go +++ b/stats/server.go @@ -22,13 +22,13 @@ func startServer(conf *config.Config) { return } - interm := map[string]interface{}{} - json.Unmarshal(first, &interm) // nolint: errcheck, gosec + interim := map[string]interface{}{} + json.Unmarshal(first, &interim) // nolint: errcheck, gosec encoder := json.NewEncoder(w) encoder.SetEscapeHTML(false) encoder.SetIndent("", " ") - if err = encoder.Encode(interm); err != nil { + if err = encoder.Encode(interim); err != nil { log.Errorw("Cannot encode json", "error", err) } }) From 3aec9657d77896835b49cea06e8639d7889fa1f6 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Wed, 17 Oct 2018 10:51:16 +0300 Subject: [PATCH 08/11] Add comments for exported functions --- stats/channels.go | 1 + stats/init.go | 1 + stats/stats.go | 2 ++ 3 files changed, 4 insertions(+) diff --git a/stats/channels.go b/stats/channels.go index bd9d9fb..f7317db 100644 --- a/stats/channels.go +++ b/stats/channels.go @@ -68,6 +68,7 @@ func EgressTraffic(traffic int) { } } +// GetStats returns a snapshot of Stats instance. func GetStats() Stats { rpcChan := make(chan Stats) statsChan <- rpcChan diff --git a/stats/init.go b/stats/init.go index 3fb206a..7b43a28 100644 --- a/stats/init.go +++ b/stats/init.go @@ -6,6 +6,7 @@ import ( "github.com/9seconds/mtg/config" ) +// Init initializes stats subsystem. func Init(conf *config.Config) error { if conf.StatsD.Enabled { client, err := newStatsd(conf) diff --git a/stats/stats.go b/stats/stats.go index e9c5999..686163d 100644 --- a/stats/stats.go +++ b/stats/stats.go @@ -89,6 +89,7 @@ func (s speed) MarshalJSON() ([]byte, error) { return json.Marshal(value) } +// Stats represents a statistics of the proxy. type Stats struct { URLs config.IPURLs `json:"urls"` Connections connections `json:"connections"` @@ -165,6 +166,7 @@ func (s *Stats) handleCrash() { s.Crashes++ } +// NewStats creates a new instance of Stats structure. func NewStats(conf *config.Config) *Stats { return &Stats{ URLs: conf.GetURLs(), From 5f00363da593756c670e189474a99cc3beabe721 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Wed, 17 Oct 2018 10:52:30 +0300 Subject: [PATCH 09/11] Add goreportcard --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index ecf09f3..6410061 100644 --- a/README.md +++ b/README.md @@ -3,6 +3,7 @@ Bullshit-free MTPROTO proxy for Telegram [![Build Status](https://travis-ci.org/9seconds/mtg.svg?branch=master)](https://travis-ci.org/9seconds/mtg) +[![Go Report Card](https://goreportcard.com/badge/github.com/9seconds/mtg)](https://goreportcard.com/report/github.com/9seconds/mtg) [![Docker Build Status](https://img.shields.io/docker/build/nineseconds/mtg.svg)](https://hub.docker.com/r/nineseconds/mtg/) # Rationale From 9346c11f37a8a8e9633dbac6a0a4c63de0f4f6b7 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Wed, 17 Oct 2018 12:06:46 +0300 Subject: [PATCH 10/11] Add support for prometheus --- config/config.go | 6 +++- go.mod | 9 +++++ go.sum | 18 ++++++++++ main.go | 8 ++++- stats/init.go | 5 +++ stats/prometheus.go | 80 +++++++++++++++++++++++++++++++++++++++++++++ stats/server.go | 2 ++ 7 files changed, 126 insertions(+), 2 deletions(-) create mode 100644 stats/prometheus.go diff --git a/config/config.go b/config/config.go index a52f76a..27f5ad8 100644 --- a/config/config.go +++ b/config/config.go @@ -38,6 +38,9 @@ type Config struct { TagsFormat statsd.TagFormat Enabled bool } + Prometheus struct { + Prefix string + } Secret []byte AdTag []byte @@ -116,7 +119,7 @@ func NewConfig(debug, verbose bool, // nolint: gocyclo bindIP, publicIPv4, publicIPv6, statsIP net.IP, bindPort, publicIPv4Port, publicIPv6Port, statsPort, statsdPort uint16, statsdIP, statsdNetwork, statsdPrefix, statsdTagsFormat string, - statsdTags map[string]string, + statsdTags map[string]string, prometheusPrefix string, secureOnly bool, secret, adtag []byte) (*Config, error) { secureMode := secureOnly @@ -174,6 +177,7 @@ func NewConfig(debug, verbose bool, // nolint: gocyclo ReadBufferSize: int(readBufferSize), WriteBufferSize: int(writeBufferSize), } + conf.Prometheus.Prefix = prometheusPrefix if statsdIP != "" { conf.StatsD.Enabled = true diff --git a/go.mod b/go.mod index 18ed04a..217fe11 100644 --- a/go.mod +++ b/go.mod @@ -4,20 +4,29 @@ require ( github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc // indirect github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf // indirect github.com/beevik/ntp v0.2.0 + github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/dustin/go-humanize v1.0.0 github.com/gofrs/uuid v3.1.0+incompatible + github.com/gogo/protobuf v1.1.1 // indirect + github.com/golang/protobuf v1.2.0 // indirect github.com/juju/errors v0.0.0-20181012004132-a4583d0a56ea github.com/juju/loggo v0.0.0-20180524022052-584905176618 // indirect github.com/juju/testing v0.0.0-20180920084828-472a3e8b2073 // indirect github.com/kr/pretty v0.1.0 // indirect + github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect github.com/pkg/errors v0.8.0 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/prometheus/client_golang v0.9.0 + github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910 // indirect + github.com/prometheus/common v0.0.0-20181015124227-bcb74de08d37 // indirect + github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d // indirect github.com/stretchr/testify v1.2.2 go.uber.org/atomic v1.3.2 // indirect go.uber.org/multierr v1.1.0 // indirect go.uber.org/zap v1.9.1 golang.org/x/net v0.0.0-20181011144130-49bb7cea24b1 // indirect + golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f // indirect gopkg.in/alecthomas/kingpin.v2 v2.2.6 gopkg.in/alexcesaro/statsd.v2 v2.0.0 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect diff --git a/go.sum b/go.sum index 33453b3..0e1b3b7 100644 --- a/go.sum +++ b/go.sum @@ -4,12 +4,18 @@ github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf h1:qet1QNfXsQxTZq github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= github.com/beevik/ntp v0.2.0 h1:sGsd+kAXzT0bfVfzJfce04g+dSRfrs+tbQW8lweuYgw= github.com/beevik/ntp v0.2.0/go.mod h1:hIHWr+l3+/clUnF44zdK+CWW7fO8dR5cIylAQ76NRpg= +github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 h1:xJ4a3vCFaGF/jqvzLMYoU8P317H5OQ+Via4RmuPwCS0= +github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo= github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= github.com/gofrs/uuid v3.1.0+incompatible h1:q2rtkjaKT4YEr6E1kamy0Ha4RtepWlQBedyHx0uzKwA= github.com/gofrs/uuid v3.1.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM= +github.com/gogo/protobuf v1.1.1 h1:72R+M5VuhED/KujmZVcIquuo8mBgX4oVda//DQb3PXo= +github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= +github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM= +github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/juju/errors v0.0.0-20181012004132-a4583d0a56ea h1:g2k+8WR7cHch4g0tBDhfiEvAp7fXxTNBiD1oC1Oxj3E= github.com/juju/errors v0.0.0-20181012004132-a4583d0a56ea/go.mod h1:W54LbzXuIE0boCoNJfwqpmkKJ1O4TCTZMetAt6jGk7Q= github.com/juju/loggo v0.0.0-20180524022052-584905176618 h1:MK144iBQF9hTSwBW/9eJm034bVoG30IshVm688T2hi8= @@ -21,10 +27,20 @@ github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORN github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU= +github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v0.9.0 h1:tXuTFVHC03mW0D+Ua1Q2d1EAVqLTuggX50V0VLICCzY= +github.com/prometheus/client_golang v0.9.0/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw= +github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910 h1:idejC8f05m9MGOsuEi1ATq9shN03HrxNkD/luQvxCv8= +github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo= +github.com/prometheus/common v0.0.0-20181015124227-bcb74de08d37 h1:Y7YdJ9Xb3MoQOzAWXnDunAJYpvhVwZdTirNfGUgPKaA= +github.com/prometheus/common v0.0.0-20181015124227-bcb74de08d37/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro= +github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d h1:GoAlyOgbOEIFdaDqxJVlbOQ1DtGmZWs/Qau0hIlk+WQ= +github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk= github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w= github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= go.uber.org/atomic v1.3.2 h1:2Oa65PReHzfn29GpvgsYwloV9AVFHPDk8tYxt2c2tr4= @@ -35,6 +51,8 @@ go.uber.org/zap v1.9.1 h1:XCJQEf3W6eZaVwhRBof6ImoYGJSITeKWsyeh3HFu/5o= go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q= golang.org/x/net v0.0.0-20181011144130-49bb7cea24b1 h1:Y/KGZSOdz/2r0WJ9Mkmz6NJBusp0kiNx1Cn82lzJQ6w= golang.org/x/net v0.0.0-20181011144130-49bb7cea24b1/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f h1:wMNYb4v58l5UBM7MYRLPG6ZhfOqbKu7X5eyFl8ZhKvA= +golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/alexcesaro/statsd.v2 v2.0.0 h1:FXkZSCZIH17vLCO5sO2UucTHsH9pc+17F6pl3JVCwMc= diff --git a/main.go b/main.go index 2d5bdcd..3a12956 100644 --- a/main.go +++ b/main.go @@ -110,6 +110,12 @@ var ( Envar("MTG_STATSD_TAGS"). StringMap() + prometheusPrefix = app.Flag("prometheus-prefix", + "Which namespace to use to send stats to Prometheus."). + Envar("MTG_PROMETHEUS_PREFIX"). + Default("mtg"). + String() + writeBufferSize = app.Flag("write-buffer", "Write buffer size in bytes. You can think about it as a buffer from client to Telegram."). Short('w'). @@ -151,7 +157,7 @@ func main() { // nolint: gocyclo *bindIP, *publicIPv4, *publicIPv6, *statsIP, *bindPort, *publicIPv4Port, *publicIPv6Port, *statsPort, *statsdPort, *statsdIP, *statsdNetwork, *statsdPrefix, *statsdTagsFormat, - *statsdTags, *secureOnly, + *statsdTags, *prometheusPrefix, *secureOnly, *secret, *adtag, ) if err != nil { diff --git a/stats/init.go b/stats/init.go index 7b43a28..ba303d7 100644 --- a/stats/init.go +++ b/stats/init.go @@ -15,6 +15,11 @@ func Init(conf *config.Config) error { } go client.run() } + prometheus, err := newPrometheus(conf) + if err != nil { + return errors.Annotate(err, "Cannot initialize prometheus client") + } + go prometheus.run() go NewStats(conf).start() go startServer(conf) diff --git a/stats/prometheus.go b/stats/prometheus.go new file mode 100644 index 0000000..00490a5 --- /dev/null +++ b/stats/prometheus.go @@ -0,0 +1,80 @@ +package stats + +import ( + "time" + + "github.com/juju/errors" + "github.com/prometheus/client_golang/prometheus" + + "github.com/9seconds/mtg/config" +) + +const prometheusPollTime = time.Second + +type prometheusExporter struct { + connections *prometheus.GaugeVec + traffic *prometheus.GaugeVec + speed *prometheus.GaugeVec + crashes prometheus.Gauge +} + +func (p *prometheusExporter) run() { + for range time.Tick(prometheusPollTime) { + instance := GetStats() + + p.connections.WithLabelValues("abridged", "v4").Set(float64(instance.Connections.Abridged.IPv4)) + p.connections.WithLabelValues("abridged", "v6").Set(float64(instance.Connections.Abridged.IPv6)) + p.connections.WithLabelValues("intermediate", "v4").Set(float64(instance.Connections.Intermediate.IPv4)) + p.connections.WithLabelValues("intermediate", "v6").Set(float64(instance.Connections.Intermediate.IPv6)) + p.connections.WithLabelValues("secure", "v4").Set(float64(instance.Connections.Secure.IPv4)) + p.connections.WithLabelValues("secure", "v6").Set(float64(instance.Connections.Secure.IPv6)) + p.traffic.WithLabelValues("ingress").Set(float64(instance.Traffic.ingress)) + p.traffic.WithLabelValues("egress").Set(float64(instance.Traffic.egress)) + p.speed.WithLabelValues("ingress").Set(float64(instance.Speed.ingress)) + p.speed.WithLabelValues("egress").Set(float64(instance.Speed.egress)) + p.crashes.Set(float64(instance.Crashes)) + } +} + +func newPrometheus(conf *config.Config) (*prometheusExporter, error) { + connections := prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: conf.Prometheus.Prefix, + Name: "connections", + Help: "Current number of connections to the proxy.", + }, []string{"type", "protocol"}) + traffic := prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: conf.Prometheus.Prefix, + Name: "traffic", + Help: "Traffic passed through the proxy in bytes.", + }, []string{"direction"}) + speed := prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: conf.Prometheus.Prefix, + Name: "speed", + Help: "Current throughput in bytes per second.", + }, []string{"direction"}) + crashes := prometheus.NewGauge(prometheus.GaugeOpts{ + Namespace: conf.Prometheus.Prefix, + Name: "crashes", + Help: "How many crashes happened.", + }) + + if err := prometheus.Register(connections); err != nil { + return nil, errors.Annotate(err, "Cannot register connections collector") + } + if err := prometheus.Register(traffic); err != nil { + return nil, errors.Annotate(err, "cannot register traffic collector") + } + if err := prometheus.Register(speed); err != nil { + return nil, errors.Annotate(err, "cannot register speed collector") + } + if err := prometheus.Register(crashes); err != nil { + return nil, errors.Annotate(err, "cannot register crashes collector") + } + + return &prometheusExporter{ + connections: connections, + traffic: traffic, + speed: speed, + crashes: crashes, + }, nil +} diff --git a/stats/server.go b/stats/server.go index b4a0fef..9d96b02 100644 --- a/stats/server.go +++ b/stats/server.go @@ -4,6 +4,7 @@ import ( "encoding/json" "net/http" + "github.com/prometheus/client_golang/prometheus/promhttp" "go.uber.org/zap" "github.com/9seconds/mtg/config" @@ -32,6 +33,7 @@ func startServer(conf *config.Config) { log.Errorw("Cannot encode json", "error", err) } }) + http.Handle("/prometheus/", promhttp.Handler()) if err := http.ListenAndServe(conf.StatAddr(), nil); err != nil { log.Fatalw("Stats server has been stopped", "error", err) From a9d0ca1e90ec724451b9c154f072f685fd310123 Mon Sep 17 00:00:00 2001 From: 9seconds Date: Wed, 17 Oct 2018 12:53:16 +0300 Subject: [PATCH 11/11] Describe and document prometheus integration --- README.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/README.md b/README.md index 6410061..d88ce2a 100644 --- a/README.md +++ b/README.md @@ -160,6 +160,7 @@ supported environment variables: | `MTG_STATSD_PREFIX` | `--statsd-prefix` | `mtg` | Which bucket prefix we should use. For example, if you set `mtg`, then metric `traffic.ingress` would be send as `mtg.traffic.ingress`. | | `MTG_STATSD_TAGS_FORMAT` | `--statsd-tags-format` | | Which tags format we should use. By default, we are using default vanilla statsd tags format but if you want to send directly to InfluxDB or Datadog, please specify it there. Possible options are `influxdb` and `datadog`. | | `MTG_STATSD_TAGS` | `--statsd-tags` | | Which tags should we send to statsd with our metrics. Please specify them as `key=value` pairs. | +| `MTG_PROMETHEUS_PREFIX` | `--prometheus-prefix` | `mtg` | Which namespace should be used for prometheus metrics. | | `MTG_BUFFER_WRITE` | `-w`, `--write-buffer` | `65536` | The size of TCP write buffer in bytes. Write buffer is the buffer for messages which are going from client to Telegram. | | `MTG_BUFFER_READ` | `-r`, `--read-buffer` | `131072` | The size of TCP read buffer in bytes. Read buffer is the buffer for messages from Telegram to client. | | `MTG_SECURE_ONLY` | `-s`, `--secure-only` | `false` | Support only clients with secure mode (i.e only clients with dd-secrets). | @@ -235,3 +236,14 @@ All metrics are gauges. Here is the list of metrics and their meaning: All metrics are prefixed with given prefix. Default prefix is `mtg`. With such prefix metric name `traffic.ingress`, for example, would be `mtg.traffic.ingress`. + + +# Prometheus integration + +[Prometheus](https://prometheus.io) integration comes out of +the box, you do not need to setup anything special. Prometheus +scrape endpoint lives on the same IP/port where generic stats +service (`http://${MTG_STATS_IP}:${MTG_STATS_PORT}`) but on +`/prometheus` path. So, if you access http stats service as `curl +http://localhost:3129/`, then your prometheus endpoint is `curl +http://localhost:3129/prometheus/`.