add tcp-pinger for measuring rtt of mailservers

Signed-off-by: Jakub Sokołowski <[email protected]>

Author: jakubgs

go.mod

@@ -27,11 +27,13 @@ require (
 	github.com/mutecomm/go-sqlcipher v0.0.0-20190227152316-55dbde17881f
 	github.com/okzk/sdnotify v0.0.0-20180710141335-d9becc38acbd
 	github.com/pborman/uuid v1.2.0
+	github.com/pkg/errors v0.8.1
 	github.com/prometheus/client_golang v1.2.1
 	github.com/russolsen/transit v0.0.0-20180705123435-0794b4c4505a
 	github.com/status-im/migrate/v4 v4.6.2-status.2
 	github.com/status-im/rendezvous v1.3.0
 	github.com/status-im/status-protocol-go v0.4.5-0.20191107122821-775d17008edf
+	github.com/status-im/tcp-shaker v0.0.0-20191113131324-2a8efc760922
 	github.com/status-im/whisper v1.5.2
 	github.com/stretchr/testify v1.4.0
 	github.com/syndtr/goleveldb v1.0.0

go.sum

@@ -578,6 +578,8 @@ github.com/status-im/rendezvous v1.3.0 h1:7RK/MXXW+tlm0asKm1u7Qp7Yni6AO29a7j8+E4
 github.com/status-im/rendezvous v1.3.0/go.mod h1:+hzjuP+j/XzLPeF6E50b88pWOTLdTcwjvNYt+Gh1W1s=
 github.com/status-im/status-protocol-go v0.4.5-0.20191107122821-775d17008edf h1:1boOd5yMePhXxYei97Rm/hFF45alUpMl87ZAWvlSKtg=
 github.com/status-im/status-protocol-go v0.4.5-0.20191107122821-775d17008edf/go.mod h1:r8TgqNOpY+fGKkBfR9PldxSSaBN0EsEEY4a3WsIh9LY=
+github.com/status-im/tcp-shaker v0.0.0-20191113131324-2a8efc760922 h1:SH100V6uGyAv5pP186c125A67AZQ+jhY77kE8Ek0lo4=
+github.com/status-im/tcp-shaker v0.0.0-20191113131324-2a8efc760922/go.mod h1:5bLqb2K4TvKZrrAY/c/d8Znmz11HnBRv2b03vkWGzzQ=
 github.com/status-im/whisper v1.5.2 h1:26NgiKusmPic38eQdtXnaY+iaQ/LuQ3Dh0kCGYT/Uxs=
 github.com/status-im/whisper v1.5.2/go.mod h1:emrOxzJme0k66QtbbQ2bdd3P8RCdLZ8sTD7SkwH1s2s=
 github.com/steakknife/bloomfilter v0.0.0-20180922174646-6819c0d2a570 h1:gIlAHnH1vJb5vwEjIp5kBj/eu99p/bl0Ay2goiPe5xE=
@@ -716,6 +718,8 @@ golang.org/x/sys v0.0.0-20190922100055-0a153f010e69/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190927073244-c990c680b611/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191010194322-b09406accb47 h1:/XfQ9z7ib8eEJX2hdgFTZJ/ntt0swNk5oYBziWeTCvY=
 golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191110163157-d32e6e3b99c4 h1:Hynbrlo6LbYI3H1IqXpkVDOcX/3HiPdhVEuyj5a59RM=
+golang.org/x/sys v0.0.0-20191110163157-d32e6e3b99c4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.0.0-20171227012246-e19ae1496984/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

rtt/rtt.go

@@ -0,0 +1,94 @@
+package rtt
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	errors "github.com/pkg/errors"
+	tcp "github.com/status-im/tcp-shaker"
+)
+
+type Result struct {
+	Address string
+	RTTMs   int
+	PingErr error
+}
+
+func runCheck(c *tcp.Checker, address string, timeout time.Duration) Result {
+	// mesaure RTT
+	start := time.Now()
+	// TCP Ping
+	err := c.CheckAddr(address, timeout)
+	// measure RTT
+	elapsed := time.Since(start)
+	latency := int(elapsed.Milliseconds())
+
+	if err != nil { // don't confuse users with valid latency values on error
+		latency = -1
+		if err == tcp.ErrTimeout {
+			err = errors.Wrap(err, "tcp check timeout")
+		} else {
+			switch err.(type) {
+			case tcp.ErrConnect:
+				err = errors.Wrap(err, "unable to connect")
+			default:
+				err = errors.Wrap(err, "unknown error")
+			}
+		}
+	}
+
+	// return results
+	return Result{
+		Address: address,
+		RTTMs:   latency,
+		PingErr: err,
+	}
+}
+
+func waitForResults(errCh <-chan error, resCh <-chan Result) (results []Result, err error) {
+	for {
+		select {
+		case err = <-errCh:
+			return nil, err
+		case res, ok := <-resCh:
+			if !ok {
+				return
+			}
+			results = append(results, res)
+		}
+	}
+}
+
+func CheckHosts(addresses []string, timeout time.Duration) ([]Result, error) {
+	c := tcp.NewChecker()
+
+	// channel for receiving possible checking loop failure
+	errCh := make(chan error, 1)
+	// channel for returning results from concurrent checks
+	resCh := make(chan Result, len(addresses))
+
+	// stop the checking loop when function exists
+	ctx, stopChecker := context.WithCancel(context.Background())
+	defer stopChecker()
+
+	// loop that queries Epoll and pipes events to CheckAddr() calls
+	go func() {
+		errCh <- c.CheckingLoop(ctx)
+	}()
+
+	var wg sync.WaitGroup
+	for i := 0; i < len(addresses); i++ {
+		wg.Add(1)
+		go func(address string, resCh chan<- Result) {
+			defer wg.Done()
+			resCh <- runCheck(c, address, timeout)
+		}(addresses[i], resCh)
+	}
+	// wait for all the routines to finish before closing results channel
+	wg.Wait()
+	close(resCh)
+
+	// wait for the results for all addresses or a checking loop error
+	return waitForResults(errCh, resCh)
+}

services/mailservers/tcp_ping.go

@@ -0,0 +1,47 @@
+package mailservers
+
+import (
+	"context"
+	"time"
+
+	"github.com/status-im/status-go/rtt"
+)
+
+type PingQuery struct {
+	Addresses []string `json:"addresses"`
+	TimeoutMs int      `json:"timeoutMs"`
+}
+
+type PingResult struct {
+	Address string  `json:"address"`
+	RTTMs   *int    `json:"latency"`
+	PingErr *string `json:"error"`
+}
+
+func toPingResult(r rtt.Result) PingResult {
+	var err *string
+	if r.PingErr != nil {
+		tmpErr := r.PingErr.Error()
+		err = &tmpErr
+	}
+	return PingResult{
+		Address: r.Address,
+		RTTMs:   &r.RTTMs,
+		PingErr: err,
+	}
+}
+
+func (a *API) Ping(ctx context.Context, pq PingQuery) ([]PingResult, error) {
+	timeout := time.Duration(pq.TimeoutMs) * time.Millisecond
+	// run the checks concurrently
+	results, err := rtt.CheckHosts(pq.Addresses, timeout)
+	if err != nil {
+		return nil, err
+	}
+	// conver to json format
+	jsonResults := make([]PingResult, len(results))
+	for i := range results {
+		jsonResults[i] = toPingResult(results[i])
+	}
+	return jsonResults, nil
+}