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-20191113234341-a6c5fb59f6ad
 	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-20191113234341-a6c5fb59f6ad h1:L0O2TPUJvbiByREFlaP4VobT/msU2/32QuUc2cnKAQU=
+github.com/status-im/tcp-shaker v0.0.0-20191113234341-a6c5fb59f6ad/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,100 @@
+package rtt
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	errors "github.com/pkg/errors"
+	tcp "github.com/status-im/tcp-shaker"
+)
+
+type Result struct {
+	Addr  string
+	RTTMs int
+	Err   error
+}
+
+// timeoutError indicates an error due to TCP connection timeout.
+// tcp-shaker returns an error implementing this interface in such a case.
+type timeoutError interface {
+	Timeout() bool
+}
+
+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
+		switch err.(type) {
+		case timeoutError:
+			err = errors.Wrap(err, "tcp check timeout")
+		case tcp.ErrConnect:
+			err = errors.Wrap(err, "unable to connect")
+		default:
+			err = err
+		}
+	}
+
+	return Result{
+		Addr:  address,
+		RTTMs: latency,
+		Err:   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)
+
+	// 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)
+	}()
+	// wait for CheckingLoop to prepare the epoll/kqueue
+	<-c.WaitReady()
+
+	// channel for returning results from concurrent checks
+	resCh := make(chan Result, len(addresses))
+
+	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,100 @@
+package mailservers
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/ethereum/go-ethereum/p2p/enode"
+	"github.com/ethereum/go-ethereum/p2p/enr"
+
+	"github.com/status-im/status-go/rtt"
+)
+
+type PingQuery struct {
+	Addresses []string `json:"addresses"`
+	TimeoutMs int      `json:"timeoutMs"`
+}
+
+type PingResult struct {
+	ENode string  `json:"address"`
+	RTTMs *int    `json:"rtt_ms"`
+	Err   *string `json:"error"`
+}
+
+func (pr *PingResult) Update(rttMs int, err error) {
+	if err != nil {
+		errStr := err.Error()
+		pr.Err = &errStr
+	}
+	pr.RTTMs = &rttMs
+}
+
+func toPingResult(r rtt.Result, addrMap map[string]string) PingResult {
+	var err *string
+	if r.Err != nil {
+		tmpErr := r.Err.Error()
+		err = &tmpErr
+	}
+	return PingResult{
+		ENode: addrMap[r.Addr],
+		RTTMs: &r.RTTMs,
+		Err:   err,
+	}
+}
+
+func parseEnodes(enodes []string) (map[string]*PingResult, []string) {
+	// parse enode addreses into normal host + port addresses
+	results := make(map[string]*PingResult, len(enodes))
+	toPing := make([]string, 0)
+
+	var ip4 enr.IPv4
+	var tcp enr.TCP
+	for i := range enodes {
+		node, err := enode.ParseV4(enodes[i])
+		if err != nil {
+			// using enode since it's irrelevant but needs to be unique
+			errStr := err.Error()
+			results[enodes[i]] = &PingResult{ENode: enodes[i], Err: &errStr}
+		}
+		// helpers for extracting values from enode.Node
+		node.Load(&ip4)
+		node.Load(&tcp)
+		addr := fmt.Sprintf("%s:%d", net.IP(ip4).String(), tcp)
+		results[addr] = &PingResult{ENode: enodes[i]}
+		toPing = append(toPing, addr)
+	}
+	return results, toPing
+}
+
+func mapValues(m map[string]*PingResult) []PingResult {
+	rval := make([]PingResult, len(m))
+	var i int
+	for _, value := range m {
+		rval[i] = *value
+		i++
+	}
+	return rval
+}
+
+func (a *API) Ping(ctx context.Context, pq PingQuery) ([]PingResult, error) {
+	timeout := time.Duration(pq.TimeoutMs) * time.Millisecond
+
+	// parse enodes into pingable addresses
+	resultsMap, toPing := parseEnodes(pq.Addresses)
+
+	// run the checks concurrently
+	results, err := rtt.CheckHosts(toPing, timeout)
+	if err != nil {
+		return nil, err
+	}
+
+	// set ping results
+	for i := range results {
+		r := results[i]
+		resultsMap[r.Addr].Update(r.RTTMs, r.Err)
+	}
+
+	return mapValues(resultsMap), nil
+}