internet-latency-collector: add aws reachability package

controlplane/internet-latency-collector/internal/awsreach/awsreach.go

@@ -0,0 +1,129 @@
+package awsreach
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"os/exec"
+	"slices"
+	"sort"
+	"time"
+
+	"github.com/malbeclabs/doublezero/controlplane/internet-latency-collector/internal/collector"
+)
+
+// PrefixesURL is served over plain HTTP; the https URL redirects back to http.
+const PrefixesURL = "http://ec2-reachability.amazonaws.com/prefixes-ipv4.json"
+
+const pingTimeout = 3 * time.Second
+
+type RegionTargets map[string][]string
+
+// ParsePrefixes keys targets by exact region name. Local Zones such as us-west-2-lax-1 carry their own key.
+func ParsePrefixes(r io.Reader) (RegionTargets, error) {
+	var raw []map[string]map[string]string
+	if err := json.NewDecoder(r).Decode(&raw); err != nil {
+		return nil, fmt.Errorf("failed to decode prefixes document: %w", err)
+	}
+	if len(raw) == 0 {
+		return nil, errors.New("prefixes document lists no regions")
+	}
+
+	parsed := make(map[string]map[string]net.IP)
+	for _, entry := range raw {
+		for region, cidrToAddress := range entry {
+			for _, candidate := range cidrToAddress {
+				ip := net.ParseIP(candidate)
+				if ip == nil || ip.To4() == nil {
+					continue
+				}
+				if parsed[region] == nil {
+					parsed[region] = make(map[string]net.IP)
+				}
+				parsed[region][ip.String()] = ip.To4()
+			}
+		}
+	}
+
+	if len(parsed) == 0 {
+		return nil, errors.New("prefixes document lists no regions with IPv4 addresses")
+	}
+
+	targets := make(RegionTargets, len(parsed))
+	for region, byAddress := range parsed {
+		ips := make([]net.IP, 0, len(byAddress))
+		for _, ip := range byAddress {
+			ips = append(ips, ip)
+		}
+		sort.Slice(ips, func(i, j int) bool { return bytes.Compare(ips[i], ips[j]) < 0 })
+
+		addresses := make([]string, len(ips))
+		for i, ip := range ips {
+			addresses[i] = ip.String()
+		}
+		targets[region] = addresses
+	}
+
+	return targets, nil
+}
+
+func FetchPrefixes(ctx context.Context, client collector.HTTPClient, url string) (RegionTargets, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create request: %w", err)
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch prefixes: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("prefixes request failed with status: %d", resp.StatusCode)
+	}
+
+	return ParsePrefixes(resp.Body)
+}
+
+type PingFunc func(ctx context.Context, address string) bool
+
+func SystemPing(ctx context.Context, address string) bool {
+	ctx, cancel := context.WithTimeout(ctx, pingTimeout)
+	defer cancel()
+
+	return exec.CommandContext(ctx, "ping", "-n", "-c", "1", "-W", "2", address).Run() == nil
+}
+
+func PreferFirst(candidates []string, preferred string) []string {
+	index := slices.Index(candidates, preferred)
+	if index <= 0 {
+		return candidates
+	}
+
+	reordered := make([]string, 0, len(candidates))
+	reordered = append(reordered, candidates[index])
+	reordered = append(reordered, candidates[:index]...)
+	reordered = append(reordered, candidates[index+1:]...)
+	return reordered
+}
+
+func FirstAnswering(ctx context.Context, candidates []string, ping PingFunc, maxAttempts int) (string, error) {
+	attempts := 0
+	for _, candidate := range candidates {
+		if attempts >= maxAttempts {
+			break
+		}
+		attempts++
+		if ping(ctx, candidate) {
+			return candidate, nil
+		}
+	}
+
+	return "", fmt.Errorf("no address answered after %d attempts", attempts)
+}

controlplane/internet-latency-collector/internal/awsreach/awsreach_test.go

@@ -0,0 +1,167 @@
+package awsreach
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+const testPrefixes = `[
+  {"us-west-2": {"35.95.0.0/16": "35.95.2.254", "34.208.0.0/12": "34.208.0.0", "35.100.0.0/16": "35.100.0.0"}},
+  {"us-west-2-lax-1": {"70.224.192.0/18": "70.224.192.0"}},
+  {"eu-central-1": {"3.64.0.0/12": "3.64.0.0", "18.153.0.0/16": "not-an-address"}}
+]`
+
+type mockHTTPClient struct {
+	DoFunc func(req *http.Request) (*http.Response, error)
+}
+
+func (m *mockHTTPClient) Do(req *http.Request) (*http.Response, error) {
+	if m.DoFunc != nil {
+		return m.DoFunc(req)
+	}
+	return nil, errors.New("mock not configured")
+}
+
+func TestInternetLatency_AWSReach_ParsePrefixes(t *testing.T) {
+	t.Parallel()
+
+	targets, err := ParsePrefixes(strings.NewReader(testPrefixes))
+	require.NoError(t, err)
+
+	require.Len(t, targets, 3, "every region key stands on its own")
+	require.Equal(t, []string{"34.208.0.0", "35.95.2.254", "35.100.0.0"}, targets["us-west-2"],
+		"addresses must be sorted numerically so output is stable")
+	require.Equal(t, []string{"70.224.192.0"}, targets["us-west-2-lax-1"],
+		"a Local Zone must not fold into its parent region")
+	require.Equal(t, []string{"3.64.0.0"}, targets["eu-central-1"],
+		"values that are not IPv4 addresses are dropped")
+}
+
+func TestInternetLatency_AWSReach_ParsePrefixes_Empty(t *testing.T) {
+	t.Parallel()
+
+	_, err := ParsePrefixes(strings.NewReader(`[]`))
+	require.Error(t, err)
+	require.ErrorContains(t, err, "no regions")
+}
+
+func TestInternetLatency_AWSReach_ParsePrefixes_NoIPv4(t *testing.T) {
+	t.Parallel()
+
+	_, err := ParsePrefixes(strings.NewReader(`[{"eu-west-1": {"18.153.0.0/16": "not-an-address"}}]`))
+	require.Error(t, err)
+	require.ErrorContains(t, err, "no regions with IPv4 addresses")
+}
+
+func TestInternetLatency_AWSReach_FetchPrefixes(t *testing.T) {
+	t.Parallel()
+
+	var requestedURL string
+	client := &mockHTTPClient{
+		DoFunc: func(req *http.Request) (*http.Response, error) {
+			requestedURL = req.URL.String()
+			return &http.Response{
+				StatusCode: http.StatusOK,
+				Body:       io.NopCloser(bytes.NewBufferString(testPrefixes)),
+			}, nil
+		},
+	}
+
+	targets, err := FetchPrefixes(t.Context(), client, PrefixesURL)
+	require.NoError(t, err)
+	require.Equal(t, "http://ec2-reachability.amazonaws.com/prefixes-ipv4.json", requestedURL,
+		"the list is served over plain HTTP only")
+	require.Contains(t, targets, "eu-central-1")
+}
+
+func TestInternetLatency_AWSReach_FetchPrefixes_BadStatus(t *testing.T) {
+	t.Parallel()
+
+	client := &mockHTTPClient{
+		DoFunc: func(req *http.Request) (*http.Response, error) {
+			return &http.Response{
+				StatusCode: http.StatusServiceUnavailable,
+				Body:       io.NopCloser(bytes.NewBufferString("")),
+			}, nil
+		},
+	}
+
+	_, err := FetchPrefixes(t.Context(), client, PrefixesURL)
+	require.Error(t, err)
+	require.ErrorContains(t, err, "status: 503")
+}
+
+func TestInternetLatency_AWSReach_PreferFirst(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name       string
+		candidates []string
+		preferred  string
+		want       []string
+	}{
+		{
+			name:       "Preferred moves to the front",
+			candidates: []string{"1.1.1.1", "2.2.2.2", "3.3.3.3"},
+			preferred:  "3.3.3.3",
+			want:       []string{"3.3.3.3", "1.1.1.1", "2.2.2.2"},
+		},
+		{
+			name:       "Preferred already first",
+			candidates: []string{"1.1.1.1", "2.2.2.2"},
+			preferred:  "1.1.1.1",
+			want:       []string{"1.1.1.1", "2.2.2.2"},
+		},
+		{
+			name:       "Preferred absent",
+			candidates: []string{"1.1.1.1", "2.2.2.2"},
+			preferred:  "9.9.9.9",
+			want:       []string{"1.1.1.1", "2.2.2.2"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			require.Equal(t, tt.want, PreferFirst(tt.candidates, tt.preferred))
+		})
+	}
+}
+
+func TestInternetLatency_AWSReach_FirstAnswering(t *testing.T) {
+	t.Parallel()
+
+	var tried []string
+	ping := func(ctx context.Context, address string) bool {
+		tried = append(tried, address)
+		return address == "3.3.3.3"
+	}
+
+	got, err := FirstAnswering(t.Context(), []string{"1.1.1.1", "2.2.2.2", "3.3.3.3", "4.4.4.4"}, ping, 8)
+	require.NoError(t, err)
+	require.Equal(t, "3.3.3.3", got)
+	require.Equal(t, []string{"1.1.1.1", "2.2.2.2", "3.3.3.3"}, tried,
+		"it stops at the first address that answers")
+}
+
+func TestInternetLatency_AWSReach_FirstAnswering_AttemptCap(t *testing.T) {
+	t.Parallel()
+
+	attempts := 0
+	ping := func(ctx context.Context, address string) bool {
+		attempts++
+		return false
+	}
+
+	_, err := FirstAnswering(t.Context(), []string{"1.1.1.1", "2.2.2.2", "3.3.3.3"}, ping, 2)
+	require.Error(t, err)
+	require.ErrorContains(t, err, "no address answered after 2 attempts")
+	require.Equal(t, 2, attempts)
+}