Faster servers information updater (#248)

* Asynchronous repeatResolve
* Parallel cyberghost and PIA (v3) processing, with a 10 goroutines limit
* Add missing vyprvpn cli flag to updater
* Increase DNS repetitions to 5 in order to obtain more IP addresses
* Update old PIA IP addresses
* Add Surfshark servers by API (unused for now)

internal/updater/cyberghost.go

@@ -27,29 +27,52 @@ func findCyberghostServers(ctx context.Context, lookupIP lookupIPFunc) (servers
 	cyberghostCountryCodes := getCyberghostSubdomainToRegion()
 	possibleCountryCodes := mergeCountryCodes(cyberghostCountryCodes, allCountryCodes)
 
+	results := make(chan models.CyberghostServer)
+	const maxGoroutines = 10
+	guard := make(chan struct{}, maxGoroutines)
 	for groupID, groupName := range groups {
 		for countryCode, region := range possibleCountryCodes {
 			if err := ctx.Err(); err != nil {
 				return nil, err
 			}
-			host := fmt.Sprintf("%s-%s.cg-dialup.net", groupID, countryCode)
-			IPs, err := resolveRepeat(ctx, lookupIP, host, 2)
-			if err != nil || len(IPs) == 0 {
-				continue
-			}
-			servers = append(servers, models.CyberghostServer{
-				Region: region,
-				Group:  groupName,
-				IPs:    IPs,
-			})
+			const domain = "cg-dialup.net"
+			host := fmt.Sprintf("%s-%s.%s", groupID, countryCode, domain)
+			guard <- struct{}{}
+			go tryCyberghostHostname(ctx, lookupIP, host, groupName, region, results)
+			<-guard
 		}
 	}
+	for i := 0; i < len(groups)*len(possibleCountryCodes); i++ {
+		server := <-results
+		if server.IPs == nil {
+			continue
+		}
+		servers = append(servers, server)
+	}
+	if err := ctx.Err(); err != nil {
+		return servers, err
+	}
 	sort.Slice(servers, func(i, j int) bool {
 		return servers[i].Region < servers[j].Region
 	})
 	return servers, nil
 }
 
+func tryCyberghostHostname(ctx context.Context, lookupIP lookupIPFunc,
+	host, groupName, region string,
+	results chan<- models.CyberghostServer) {
+	IPs, err := resolveRepeat(ctx, lookupIP, host, 2)
+	if err != nil || len(IPs) == 0 {
+		results <- models.CyberghostServer{}
+		return
+	}
+	results <- models.CyberghostServer{
+		Region: region,
+		Group:  groupName,
+		IPs:    IPs,
+	}
+}
+
 //nolint:goconst
 func stringifyCyberghostServers(servers []models.CyberghostServer) (s string) {
 	s = "func CyberghostServers() []models.CyberghostServer {\n"

internal/updater/pia.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"sort"
 	"strings"
+	"sync"
 
 	"github.com/qdm12/gluetun/internal/models"
 )
@@ -50,11 +51,21 @@ func (u *updater) updatePIAOld(ctx context.Context) (err error) {
 	if err != nil {
 		return err
 	}
+	const maxGoroutines = 10
+	guard := make(chan struct{}, maxGoroutines)
+	errors := make(chan error)
+	serversCh := make(chan models.PIAServer)
 	servers := make([]models.PIAServer, 0, len(contents))
+	ctx, cancel := context.WithCancel(ctx)
+	wg := &sync.WaitGroup{}
+	defer func() {
+		cancel()
+		wg.Wait()
+		defer close(guard)
+		defer close(errors)
+		defer close(serversCh)
+	}()
 	for fileName, content := range contents {
-		if err := ctx.Err(); err != nil {
-			return err
-		}
 		remoteLines := extractRemoteLinesFromOpenvpn(content)
 		if len(remoteLines) == 0 {
 			return fmt.Errorf("cannot find any remote lines in %s", fileName)
@@ -63,20 +74,19 @@ func (u *updater) updatePIAOld(ctx context.Context) (err error) {
 		if len(hosts) == 0 {
 			return fmt.Errorf("cannot find any hosts in %s", fileName)
 		}
-		var IPs []net.IP
-		for _, host := range hosts {
-			newIPs, err := resolveRepeat(ctx, u.lookupIP, host, 3)
-			if err != nil {
-				return err
-			}
-			IPs = append(IPs, newIPs...)
-		}
 		region := strings.TrimSuffix(fileName, ".ovpn")
-		server := models.PIAServer{
-			Region: region,
-			IPs:    uniqueSortedIPs(IPs),
+		guard <- struct{}{}
+		wg.Add(1)
+		go resolvePIAHostname(ctx, wg, region, hosts, u.lookupIP, errors, serversCh)
+		<-guard
+	}
+	for range contents {
+		select {
+		case err := <-errors:
+			return err
+		case server := <-serversCh:
+			servers = append(servers, server)
 		}
-		servers = append(servers, server)
 	}
 	sort.Slice(servers, func(i, j int) bool {
 		return servers[i].Region < servers[j].Region
@@ -89,6 +99,28 @@ func (u *updater) updatePIAOld(ctx context.Context) (err error) {
 	return nil
 }
 
+func resolvePIAHostname(ctx context.Context, wg *sync.WaitGroup,
+	region string, hosts []string, lookupIP lookupIPFunc,
+	errors chan<- error, serversCh chan<- models.PIAServer) {
+	defer wg.Done()
+	var IPs []net.IP //nolint:prealloc
+	// usually one single host in this case
+	// so no need to run in goroutines the for loop below
+	for _, host := range hosts {
+		const repetition = 5
+		newIPs, err := resolveRepeat(ctx, lookupIP, host, repetition)
+		if err != nil {
+			errors <- err
+			return
+		}
+		IPs = append(IPs, newIPs...)
+	}
+	serversCh <- models.PIAServer{
+		Region: region,
+		IPs:    uniqueSortedIPs(IPs),
+	}
+}
+
 func stringifyPIAServers(servers []models.PIAServer) (s string) {
 	s = "func PIAServers() []models.PIAServer {\n"
 	s += "	return []models.PIAServer{\n"

internal/updater/purevpn.go

@@ -90,7 +90,7 @@ func findPurevpnServers(ctx context.Context, httpGet httpGetFunc, lookupIP looku
 			continue
 		}
 		host := jsonServer.UDP
-		const repetition = 3
+		const repetition = 5
 		IPs, err := resolveRepeat(ctx, lookupIP, host, repetition)
 		if err != nil {
 			warnings = append(warnings, err.Error())

internal/updater/resolver.go

@@ -1,8 +1,10 @@
 package updater
 
 import (
+	"bytes"
 	"context"
 	"net"
+	"sort"
 )
 
 func newResolver(resolverAddress string) *net.Resolver {
@@ -30,12 +32,49 @@ func newLookupIP(r *net.Resolver) lookupIPFunc {
 }
 
 func resolveRepeat(ctx context.Context, lookupIP lookupIPFunc, host string, n int) (ips []net.IP, err error) {
+	foundIPs := make(chan []net.IP)
+	errors := make(chan error)
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
 	for i := 0; i < n; i++ {
-		newIPs, err := lookupIP(ctx, host)
-		if err != nil {
-			return nil, err
-		}
-		ips = append(ips, newIPs...)
+		go func() {
+			newIPs, err := lookupIP(ctx, host)
+			if err != nil {
+				errors <- err
+			} else {
+				foundIPs <- newIPs
+			}
+		}()
 	}
-	return uniqueSortedIPs(ips), nil
+
+	uniqueIPs := make(map[string]struct{})
+	for i := 0; i < n; i++ {
+		select {
+		case newIPs := <-foundIPs:
+			for _, ip := range newIPs {
+				key := ip.String()
+				uniqueIPs[key] = struct{}{}
+			}
+		case newErr := <-errors:
+			if err == nil {
+				err = newErr
+				cancel()
+			}
+		}
+	}
+
+	ips = make([]net.IP, 0, len(uniqueIPs))
+	for key := range uniqueIPs {
+		ip := net.ParseIP(key)
+		if ipv4 := ip.To4(); ipv4 != nil {
+			ip = ipv4
+		}
+		ips = append(ips, ip)
+	}
+
+	sort.Slice(ips, func(i, j int) bool {
+		return bytes.Compare(ips[i], ips[j]) < 1
+	})
+
+	return ips, err
 }

internal/updater/resolver_test.go

@@ -0,0 +1,75 @@
+package updater
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_resolveRepeat(t *testing.T) {
+	t.Parallel()
+	testCases := map[string]struct {
+		lookupIPResult [][]net.IP
+		lookupIPErr    error
+		n              int
+		ips            []net.IP
+		err            error
+	}{
+		"failure": {
+			lookupIPResult: [][]net.IP{
+				{{1, 1, 1, 1}, {1, 1, 1, 2}},
+			},
+			lookupIPErr: fmt.Errorf("feeling sick"),
+			n:           1,
+			ips:         []net.IP{},
+			err:         fmt.Errorf("feeling sick"),
+		},
+		"successful": {
+			lookupIPResult: [][]net.IP{
+				{{1, 1, 1, 1}, {1, 1, 1, 2}},
+				{{2, 1, 1, 1}, {2, 1, 1, 2}},
+				{{2, 1, 1, 3}, {2, 1, 1, 2}},
+			},
+			n: 3,
+			ips: []net.IP{
+				{1, 1, 1, 1},
+				{1, 1, 1, 2},
+				{2, 1, 1, 1},
+				{2, 1, 1, 2},
+				{2, 1, 1, 3},
+			},
+		},
+	}
+	for name, testCase := range testCases {
+		testCase := testCase
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			if testCase.lookupIPErr == nil {
+				require.Len(t, testCase.lookupIPResult, testCase.n)
+			}
+			const host = "blabla"
+			i := 0
+			lookupIP := func(ctx context.Context, argHost string) (
+				ips []net.IP, err error) {
+				assert.Equal(t, host, argHost)
+				result := testCase.lookupIPResult[i]
+				i++
+				return result, testCase.err
+			}
+
+			ips, err := resolveRepeat(
+				context.Background(), lookupIP, host, testCase.n)
+			if testCase.err != nil {
+				require.Error(t, err)
+				assert.Equal(t, testCase.err.Error(), err.Error())
+			} else {
+				assert.NoError(t, err)
+			}
+			assert.Equal(t, testCase.ips, ips)
+		})
+	}
+}

internal/updater/surfshark.go

@@ -2,7 +2,9 @@ package updater
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"net"
 	"sort"
 	"strings"
@@ -11,7 +13,7 @@ import (
 )
 
 func (u *updater) updateSurfshark(ctx context.Context) (err error) {
-	servers, warnings, err := findSurfsharkServers(ctx, u.lookupIP)
+	servers, warnings, err := findSurfsharkServersFromZip(ctx, u.lookupIP)
 	if u.options.CLI {
 		for _, warning := range warnings {
 			u.logger.Warn("Surfshark: %s", warning)
@@ -28,7 +30,47 @@ func (u *updater) updateSurfshark(ctx context.Context) (err error) {
 	return nil
 }
 
-func findSurfsharkServers(ctx context.Context, lookupIP lookupIPFunc) (servers []models.SurfsharkServer, warnings []string, err error) {
+//nolint:deadcode,unused
+func findSurfsharkServersFromAPI(ctx context.Context, lookupIP lookupIPFunc, httpGet httpGetFunc) (servers []models.SurfsharkServer, warnings []string, err error) {
+	const url = "https://my.surfshark.com/vpn/api/v1/server/clusters"
+	response, err := httpGet(url)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer response.Body.Close()
+	b, err := ioutil.ReadAll(response.Body)
+	if err != nil {
+		return nil, nil, err
+	}
+	var jsonServers []struct {
+		Host     string `json:"connectionName"`
+		Country  string `json:"country"`
+		Location string `json:"location"`
+	}
+	if err := json.Unmarshal(b, &jsonServers); err != nil {
+		return nil, nil, err
+	}
+	for _, jsonServer := range jsonServers {
+		host := jsonServer.Host
+		const repetition = 5
+		IPs, err := resolveRepeat(ctx, lookupIP, host, repetition)
+		if err != nil {
+			return nil, warnings, err
+		} else if len(IPs) == 0 {
+			warning := fmt.Sprintf("no IP address found for host %q", host)
+			warnings = append(warnings, warning)
+			continue
+		}
+		server := models.SurfsharkServer{
+			Region: jsonServer.Country + " " + jsonServer.Location,
+			IPs:    uniqueSortedIPs(IPs),
+		}
+		servers = append(servers, server)
+	}
+	return servers, warnings, nil
+}
+
+func findSurfsharkServersFromZip(ctx context.Context, lookupIP lookupIPFunc) (servers []models.SurfsharkServer, warnings []string, err error) {
 	const zipURL = "https://my.surfshark.com/vpn/api/v1/server/configurations"
 	contents, err := fetchAndExtractFiles(zipURL)
 	if err != nil {
@@ -97,6 +139,7 @@ func findSurfsharkServers(ctx context.Context, lookupIP lookupIPFunc) (servers [
 	})
 	return servers, warnings, nil
 }
+
 func getRemainingServers(ctx context.Context, mapping map[string]string, lookupIP lookupIPFunc) (
 	servers []models.SurfsharkServer, warnings []string, err error) {
 	for subdomain, region := range mapping {

internal/updater/vyprvpn.go

@@ -43,7 +43,8 @@ func findVyprvpnServers(ctx context.Context, lookupIP lookupIPFunc) (servers []m
 		}
 		var IPs []net.IP
 		for _, host := range hosts {
-			newIPs, err := lookupIP(ctx, host)
+			const repetitions = 1
+			newIPs, err := resolveRepeat(ctx, lookupIP, host, repetitions)
 			if err != nil {
 				return nil, err
 			}

internal/updater/windscribe.go

@@ -31,7 +31,8 @@ func findWindscribeServers(ctx context.Context, lookupIP lookupIPFunc) (servers
 			return nil, err
 		}
 		host := countryCode + "." + domain
-		ips, err := resolveRepeat(ctx, lookupIP, host, 2)
+		const repetitions = 5
+		ips, err := resolveRepeat(ctx, lookupIP, host, repetitions)
 		if err != nil || len(ips) == 0 {
 			continue
 		}